enable source connector to set ForwardMessageProperty

api/v1alpha1/source_types.go

@@ -40,14 +40,14 @@ type SourceSpec struct {
 
 	// +kubebuilder:validation:Optional
 	// +kubebuilder:pruning:PreserveUnknownFields
-	SourceConfig        *Config                     `json:"sourceConfig,omitempty"`
-	Resources           corev1.ResourceRequirements `json:"resources,omitempty"`
-	SecretsMap          map[string]SecretRef        `json:"secretsMap,omitempty"`
-	ProcessingGuarantee ProcessGuarantee            `json:"processingGuarantee,omitempty"`
-	RuntimeFlags        string                      `json:"runtimeFlags,omitempty"`
-	VolumeMounts        []corev1.VolumeMount        `json:"volumeMounts,omitempty"`
-
-	Pod PodPolicy `json:"pod,omitempty"`
+	SourceConfig                 *Config                     `json:"sourceConfig,omitempty"`
+	Resources                    corev1.ResourceRequirements `json:"resources,omitempty"`
+	SecretsMap                   map[string]SecretRef        `json:"secretsMap,omitempty"`
+	ProcessingGuarantee          ProcessGuarantee            `json:"processingGuarantee,omitempty"`
+	RuntimeFlags                 string                      `json:"runtimeFlags,omitempty"`
+	VolumeMounts                 []corev1.VolumeMount        `json:"volumeMounts,omitempty"`
+	ForwardSourceMessageProperty *bool                       `json:"forwardSourceMessageProperty,omitempty"`
+	Pod                          PodPolicy                   `json:"pod,omitempty"`
 
 	Messaging `json:",inline"`
 	Runtime   `json:",inline"`

api/v1alpha1/source_webhook.go

@@ -82,6 +82,11 @@ func (r *Source) Default() {
 		}
 	}
 
+	if r.Spec.ForwardSourceMessageProperty == nil {
+		trueVal := true
+		r.Spec.ForwardSourceMessageProperty = &trueVal
+	}
+
 	if r.Spec.Output.ProducerConf == nil {
 		producerConf := &ProducerConfig{
 			MaxPendingMessages:                 1000,

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

@@ -4450,6 +4450,8 @@ spec:
                     type: string
                   clusterName:
                     type: string
+                  forwardSourceMessageProperty:
+                    type: boolean
                   golang:
                     properties:
                       go:

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

@@ -36,6 +36,8 @@ spec:
               type: string
             clusterName:
               type: string
+            forwardSourceMessageProperty:
+              type: boolean
             golang:
               properties:
                 go:

config/samples/compute_v1alpha1_source.yaml

@@ -13,6 +13,7 @@ spec:
       useThreadLocalProducers: true
     topic: persistent://public/default/destination
     typeClassName: org.apache.pulsar.common.schema.KeyValue
+  forwardSourceMessageProperty: true
   resources:
     limits:
       cpu: "0.2"

controllers/spec/utils.go

@@ -245,12 +245,17 @@ func generateSourceOutputSpec(source *v1alpha1.Source) *proto.SinkSpec {
 			BatchBuilder:                       source.Spec.Output.ProducerConf.BatchBuilder,
 		}
 	}
+	var forward = false
+	if source.Spec.ForwardSourceMessageProperty != nil {
+		forward = *source.Spec.ForwardSourceMessageProperty
+	}
 	return &proto.SinkSpec{
-		TypeClassName:  source.Spec.Output.TypeClassName,
-		Topic:          source.Spec.Output.Topic,
-		ProducerSpec:   &producerSpec,
-		SerDeClassName: source.Spec.Output.SinkSerdeClassName,
-		SchemaType:     source.Spec.Output.SinkSchemaType,
+		TypeClassName:                source.Spec.Output.TypeClassName,
+		Topic:                        source.Spec.Output.Topic,
+		ProducerSpec:                 &producerSpec,
+		SerDeClassName:               source.Spec.Output.SinkSerdeClassName,
+		SchemaType:                   source.Spec.Output.SinkSchemaType,
+		ForwardSourceMessageProperty: forward,
 	}
 }
 

manifests/crd.yaml

@@ -4448,6 +4448,8 @@ spec:
                     type: string
                   clusterName:
                     type: string
+                  forwardSourceMessageProperty:
+                    type: boolean
                   golang:
                     properties:
                       go:
@@ -11212,6 +11214,8 @@ spec:
               type: string
             clusterName:
               type: string
+            forwardSourceMessageProperty:
+              type: boolean
             golang:
               properties:
                 go:

mesh-worker-service/integration-tests/docker/Dockerfile

@@ -1,3 +1,22 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
 FROM streamnative/pulsar-all:2.8.0.7
 COPY ./target/mesh-worker-service*.nar /pulsar/mesh-worker-service.nar
 COPY ./integration-tests/docker/connectors.yaml /pulsar/conf/connectors.yaml

mesh-worker-service/integration-tests/docker/connectors.yaml

@@ -16,6 +16,7 @@
 # specific language governing permissions and limitations
 # under the License.
 #
+
 - id: pulsar-io-data-generator
   name: data-generator
   description: Test data generator connector

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

@@ -131,6 +131,7 @@ public static V1alpha1Source createV1alpha1SourceFromSourceConfig(String kind, S
         if (Strings.isNotEmpty(functionDetails.getSink().getSchemaType())) {
             v1alpha1SourceSpecOutput.setSinkSchemaType(functionDetails.getSink().getSchemaType());
         }
+        v1alpha1SourceSpec.setForwardSourceMessageProperty(functionDetails.getSink().getForwardSourceMessageProperty());
         // process ProducerConf
         V1alpha1SourceSpecOutputProducerConf v1alpha1SourceSpecOutputProducerConf
                 = new V1alpha1SourceSpecOutputProducerConf();