#192 add defaultSchemaType and defaultSerdeClassName to FunctionMeshConnectorDefinition (#195)

* add defaultSchemaType and defaultSerdeClassName to FunctionMeshConnectorDefinition

* check if update

Author: freeznet

mesh-worker-service/src/main/java/io/functionmesh/compute/models/FunctionMeshConnectorDefinition.java

@@ -63,6 +63,16 @@ public class FunctionMeshConnectorDefinition extends ConnectorDefinition {
      */
     private String typeClassName;
 
+    /**
+     * Default schema type of the connector's topic, optional.
+     */
+    private String defaultSchemaType;
+
+    /**
+     * Default serde class name of the connector's topic, optional.
+     */
+    private String defaultSerdeClassName;
+
     public String toFullImageURL() {
         return String.format("%s%s:%s", imageRegistry != null ? imageRegistry : DEFAULT_REGISTRY,
                 imageRepository, imageTag != null ? imageTag : version);

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

@@ -160,6 +160,31 @@ public static V1alpha1Sink createV1alpha1SkinFromSinkConfig(String kind, String
                     } else {
                         v1alpha1SinkSpecInput.setTypeClassName(functionMeshConnectorDefinition.getTypeClassName());
                     }
+                    // we only handle user provide --inputs but also with defaultSchemaType defined
+                    if (sinkConfig.getInputs() != null && sinkConfig.getInputs().size() > 0) {
+                        if (v1alpha1SinkSpecInput.getSourceSpecs() == null) {
+                            v1alpha1SinkSpecInput.setSourceSpecs(new HashMap<>());
+                        }
+                        for (String input : sinkConfig.getInputs()) {
+                            V1alpha1SinkSpecInputSourceSpecs inputSourceSpecsItem =
+                                    v1alpha1SinkSpecInput.getSourceSpecs().getOrDefault(input,
+                                            new V1alpha1SinkSpecInputSourceSpecs());
+                            boolean updated = false;
+                            if (StringUtils.isNotEmpty(functionMeshConnectorDefinition.getDefaultSchemaType())
+                                    && StringUtils.isEmpty(inputSourceSpecsItem.getSchemaType())) {
+                                inputSourceSpecsItem.setSchemaType(functionMeshConnectorDefinition.getDefaultSchemaType());
+                                updated = true;
+                            }
+                            if (StringUtils.isNotEmpty(functionMeshConnectorDefinition.getDefaultSerdeClassName())
+                                    && StringUtils.isEmpty(inputSourceSpecsItem.getSerdeClassname())) {
+                                inputSourceSpecsItem.setSerdeClassname(functionMeshConnectorDefinition.getDefaultSerdeClassName());
+                                updated = true;
+                            }
+                            if (updated) {
+                                v1alpha1SinkSpecInput.putSourceSpecsItem(input, inputSourceSpecsItem);
+                            }
+                        }
+                    }
                 }
             }
         }

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

@@ -32,10 +32,8 @@
 import io.functionmesh.compute.worker.MeshConnectorsManager;
 import io.kubernetes.client.custom.Quantity;
 import lombok.extern.slf4j.Slf4j;
-import io.functionmesh.compute.models.CustomRuntimeOptions;
 import org.apache.commons.lang.StringUtils;
 import org.apache.logging.log4j.util.Strings;
-import org.apache.pulsar.common.functions.FunctionDefinition;
 import org.apache.pulsar.common.functions.ProducerConfig;
 import org.apache.pulsar.common.functions.Resources;
 import org.apache.pulsar.common.io.SourceConfig;
@@ -44,9 +42,7 @@
 import org.apache.pulsar.functions.utils.SourceConfigUtils;
 
 import javax.ws.rs.core.Response;
-import java.io.IOException;
 import java.io.InputStream;
-import java.net.URISyntaxException;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -165,6 +161,15 @@ public static V1alpha1Source createV1alpha1SourceFromSourceConfig(String kind, S
                     } else {
                         v1alpha1SourceSpecOutput.setTypeClassName(functionMeshConnectorDefinition.getTypeClassName());
                     }
+                    // use default schema type if user not provided
+                    if (StringUtils.isNotEmpty(functionMeshConnectorDefinition.getDefaultSchemaType())
+                            && StringUtils.isEmpty(v1alpha1SourceSpecOutput.getSinkSchemaType())) {
+                        v1alpha1SourceSpecOutput.setSinkSchemaType(functionMeshConnectorDefinition.getDefaultSchemaType());
+                    }
+                    if (StringUtils.isNotEmpty(functionMeshConnectorDefinition.getDefaultSerdeClassName())
+                            && StringUtils.isEmpty(v1alpha1SourceSpecOutput.getSinkSerdeClassName())) {
+                        v1alpha1SourceSpecOutput.setSinkSerdeClassName(functionMeshConnectorDefinition.getDefaultSerdeClassName());
+                    }
                 }
             }
         }