remove logTopic from connector CRD (#242)

* remove logTopic from connector CRD

* remove logTopic from mesh-worker-service

* remove logTopic

Author: freeznet

api/v1alpha1/sink_types.go

@@ -37,7 +37,6 @@ type SinkSpec struct {
 	Replicas     *int32                      `json:"replicas,omitempty"`
 	MaxReplicas  *int32                      `json:"maxReplicas,omitempty"` // if provided, turn on autoscaling
 	Input        InputConf                   `json:"input,omitempty"`
-	LogTopic     string                      `json:"logTopic,omitempty"`
 	SinkConfig   map[string]string           `json:"sinkConfig,omitempty"`
 	Resources    corev1.ResourceRequirements `json:"resources,omitempty"`
 	SecretsMap   map[string]SecretRef        `json:"secretsMap,omitempty"`

api/v1alpha1/source_types.go

@@ -37,7 +37,6 @@ type SourceSpec struct {
 	Replicas            *int32                      `json:"replicas,omitempty"`
 	MaxReplicas         *int32                      `json:"maxReplicas,omitempty"` // if provided, turn on autoscaling
 	Output              OutputConf                  `json:"output,omitempty"`
-	LogTopic            string                      `json:"logTopic,omitempty"`
 	SourceConfig        map[string]string           `json:"sourceConfig,omitempty"`
 	Resources           corev1.ResourceRequirements `json:"resources,omitempty"`
 	SecretsMap          map[string]SecretRef        `json:"secretsMap,omitempty"`

config/crd/bases/compute.functionmesh.io_functionmeshes.yaml

@@ -2373,8 +2373,6 @@ spec:
                       jarLocation:
                         type: string
                     type: object
-                  logTopic:
-                    type: string
                   maxMessageRetry:
                     format: int32
                     type: integer
@@ -4474,8 +4472,6 @@ spec:
                       jarLocation:
                         type: string
                     type: object
-                  logTopic:
-                    type: string
                   maxReplicas:
                     format: int32
                     type: integer

config/crd/bases/compute.functionmesh.io_sinks.yaml

@@ -134,8 +134,6 @@ spec:
                 jarLocation:
                   type: string
               type: object
-            logTopic:
-              type: string
             maxMessageRetry:
               format: int32
               type: integer

config/crd/bases/compute.functionmesh.io_sources.yaml

@@ -56,8 +56,6 @@ spec:
                 jarLocation:
                   type: string
               type: object
-            logTopic:
-              type: string
             maxReplicas:
               format: int32
               type: integer

controllers/spec/utils.go

@@ -207,7 +207,6 @@ func convertSourceDetails(source *v1alpha1.Source) *proto.FunctionDetails {
 		Namespace:            source.Namespace,
 		Name:                 source.Name,
 		ClassName:            "org.apache.pulsar.functions.api.utils.IdentityFunction",
-		LogTopic:             source.Spec.LogTopic,
 		ProcessingGuarantees: convertProcessingGuarantee(source.Spec.ProcessingGuarantee),
 		UserConfig:           getUserConfig(source.Spec.SourceConfig),
 		SecretsMap:           marshalSecretsMap(source.Spec.SecretsMap),
@@ -261,7 +260,6 @@ func convertSinkDetails(sink *v1alpha1.Sink) *proto.FunctionDetails {
 		Namespace:            sink.Namespace,
 		Name:                 sink.Name,
 		ClassName:            "org.apache.pulsar.functions.api.utils.IdentityFunction",
-		LogTopic:             sink.Spec.LogTopic,
 		ProcessingGuarantees: convertProcessingGuarantee(sink.Spec.ProcessingGuarantee),
 		SecretsMap:           marshalSecretsMap(sink.Spec.SecretsMap),
 		Runtime:              proto.FunctionDetails_JAVA,

manifests/crd.yaml

@@ -2371,8 +2371,6 @@ spec:
                       jarLocation:
                         type: string
                     type: object
-                  logTopic:
-                    type: string
                   maxMessageRetry:
                     format: int32
                     type: integer
@@ -4472,8 +4470,6 @@ spec:
                       jarLocation:
                         type: string
                     type: object
-                  logTopic:
-                    type: string
                   maxReplicas:
                     format: int32
                     type: integer
@@ -9078,8 +9074,6 @@ spec:
                 jarLocation:
                   type: string
               type: object
-            logTopic:
-              type: string
             maxMessageRetry:
               format: int32
               type: integer
@@ -11243,8 +11237,6 @@ spec:
                 jarLocation:
                   type: string
               type: object
-            logTopic:
-              type: string
             maxReplicas:
               format: int32
               type: integer

mesh-worker-service/src/main/java/io/functionmesh/compute/util/SinksUtil.java

@@ -219,10 +219,6 @@ public static V1alpha1Sink createV1alpha1SkinFromSinkConfig(String kind, String
             v1alpha1SinkSpec.setTimeout((int) functionDetails.getSource().getTimeoutMs());
         }
 
-        if (Strings.isNotEmpty(functionDetails.getLogTopic())) {
-            v1alpha1SinkSpec.setLogTopic(functionDetails.getLogTopic());
-        }
-
         if (functionDetails.hasRetryDetails()) {
             v1alpha1SinkSpec.setMaxMessageRetry(functionDetails.getRetryDetails().getMaxMessageRetries());
             if (Strings.isNotEmpty(functionDetails.getRetryDetails().getDeadLetterTopic())) {

mesh-worker-service/src/main/java/io/functionmesh/compute/util/SourcesUtil.java

@@ -181,10 +181,6 @@ public static V1alpha1Source createV1alpha1SourceFromSourceConfig(String kind, S
 
         v1alpha1SourceSpec.setOutput(v1alpha1SourceSpecOutput);
 
-        if (!org.apache.commons.lang3.StringUtils.isEmpty(functionDetails.getLogTopic())) {
-            v1alpha1SourceSpec.setLogTopic(functionDetails.getLogTopic());
-        }
-
         v1alpha1SourceSpec.setReplicas(functionDetails.getParallelism());
         if (customRuntimeOptions.getMaxReplicas() > functionDetails.getParallelism()) {
             v1alpha1SourceSpec.setMaxReplicas(customRuntimeOptions.getMaxReplicas());