[issue apache#6765] Expose definition flags to function (apache#6868)

1）
Change the value of the CLI tool parameter "--custom-schema-input" from
"TopicName-> schemaType" 
 to
"topicName-> {" schemaType ":" type"," jsr310ConversionEnabled ": true," alwaysAllowNull ": true}"

2）
Modify Function.proto, add properties "jsr310ConversionEnabled", "alwaysAllowNull"。So that we can receive the above 2 parameters

3）
Modify the "FunctionConfigUtils#convert" method ， put the two parameters in "CustomSchemaInputs" into ConsumerSpec

4）
In “JavaInstanceRunnable#setupInput” method, put the above 2 parameters into "ConsumerConfig" and pass it to "PulsarSource", let it set the parameters into "schema" when creating the consumer of Source。
So that，The “function” can get these 2 parameters from the message of 
”currentRecord“

pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java

@@ -59,6 +59,8 @@
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.SchemaDefinition;
 import org.apache.pulsar.client.impl.auth.AuthenticationTls;
 import org.apache.pulsar.common.functions.ConsumerConfig;
 import org.apache.pulsar.common.functions.FunctionConfig;
@@ -72,6 +74,7 @@
 import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
 import org.apache.pulsar.functions.LocalRunner;
+import org.apache.pulsar.functions.api.examples.pojo.AvroTestObject;
 import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactory;
 import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactoryConfig;
 import org.apache.pulsar.io.datagenerator.DataGeneratorPrintSink;
@@ -479,11 +482,134 @@ private void testE2EPulsarFunctionLocalRun(String jarFilePathUrl) throws Excepti
         }
     }
 
+    public void testAvroFunctionLocalRun(String jarFilePathUrl) throws Exception {
+
+        final String namespacePortion = "io";
+        final String replNamespace = tenant + "/" + namespacePortion;
+        final String sourceTopic = "persistent://" + replNamespace + "/my-topic1";
+        final String sinkTopic = "persistent://" + replNamespace + "/output";
+        final String propertyKey = "key";
+        final String propertyValue = "value";
+        final String functionName = "PulsarFunction-test";
+        final String subscriptionName = "test-sub";
+        admin.namespaces().createNamespace(replNamespace);
+        Set<String> clusters = Sets.newHashSet(Lists.newArrayList(CLUSTER));
+        admin.namespaces().setNamespaceReplicationClusters(replNamespace, clusters);
+
+
+        Schema schema = Schema.AVRO(SchemaDefinition.builder()
+                .withAlwaysAllowNull(true)
+                .withJSR310ConversionEnabled(true)
+                .withPojo(AvroTestObject.class).build());
+        //use AVRO schema
+        admin.schemas().createSchema(sourceTopic, schema.getSchemaInfo());
+
+        //produce message to sourceTopic
+        Producer<AvroTestObject> producer = pulsarClient.newProducer(schema).topic(sourceTopic).create();
+        //consume message from sinkTopic
+        Consumer<GenericRecord> consumer = pulsarClient.newConsumer(Schema.AUTO_CONSUME()).topic(sinkTopic).subscriptionName("sub").subscribe();
+
+        FunctionConfig functionConfig = createFunctionConfig(tenant, namespacePortion, functionName,
+                sourceTopic, sinkTopic, subscriptionName);
+        //set jsr310ConversionEnabled、alwaysAllowNull
+        Map<String,String> schemaInput = new HashMap<>();
+        schemaInput.put(sourceTopic, "{\"schemaType\":\"AVRO\",\"schemaProperties\":{\"__jsr310ConversionEnabled\":\"true\",\"__alwaysAllowNull\":\"true\"}}");
+        Map<String, String> schemaOutput = new HashMap<>();
+        schemaOutput.put(sinkTopic, "{\"schemaType\":\"AVRO\",\"schemaProperties\":{\"__jsr310ConversionEnabled\":\"true\",\"__alwaysAllowNull\":\"true\"}}");
+
+        functionConfig.setCustomSchemaInputs(schemaInput);
+        functionConfig.setCustomSchemaOutputs(schemaOutput);
+        functionConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE);
+        if (jarFilePathUrl == null) {
+            functionConfig.setClassName("org.apache.pulsar.functions.api.examples.AvroSchemaTestFunction");
+        } else {
+            functionConfig.setJar(jarFilePathUrl);
+        }
+
+        LocalRunner localRunner = LocalRunner.builder()
+                .functionConfig(functionConfig)
+                .clientAuthPlugin(AuthenticationTls.class.getName())
+                .clientAuthParams(String.format("tlsCertFile:%s,tlsKeyFile:%s", TLS_CLIENT_CERT_FILE_PATH, TLS_CLIENT_KEY_FILE_PATH))
+                .useTls(true)
+                .tlsTrustCertFilePath(TLS_TRUST_CERT_FILE_PATH)
+                .tlsAllowInsecureConnection(true)
+                .tlsHostNameVerificationEnabled(false)
+                .brokerServiceUrl(pulsar.getBrokerServiceUrlTls()).build();
+        localRunner.start(false);
+
+        retryStrategically((test) -> {
+            try {
+                TopicStats stats = admin.topics().getStats(sourceTopic);
+                return stats.subscriptions.get(subscriptionName) != null
+                        && !stats.subscriptions.get(subscriptionName).consumers.isEmpty();
+            } catch (PulsarAdminException e) {
+                return false;
+            }
+        }, 50, 150);
+
+        int totalMsgs = 5;
+        for (int i = 0; i < totalMsgs; i++) {
+            AvroTestObject avroTestObject = new AvroTestObject();
+            avroTestObject.setBaseValue(i);
+            producer.newMessage().property(propertyKey, propertyValue)
+                    .value(avroTestObject).send();
+        }
+
+        //consume message from sinkTopic
+        for (int i = 0; i < totalMsgs; i++) {
+            Message<GenericRecord> msg = consumer.receive(5, TimeUnit.SECONDS);
+            String receivedPropertyValue = msg.getProperty(propertyKey);
+            assertEquals(propertyValue, receivedPropertyValue);
+            assertEquals(msg.getValue().getField("baseValue"),  10 + i);
+            consumer.acknowledge(msg);
+        }
+
+        // validate pulsar-sink consumer has consumed all messages
+        assertNotEquals(admin.topics().getStats(sinkTopic).subscriptions.values().iterator().next().unackedMessages, 0);
+        localRunner.stop();
+
+        retryStrategically((test) -> {
+            try {
+                TopicStats topicStats = admin.topics().getStats(sourceTopic);
+                return topicStats.subscriptions.get(subscriptionName) != null
+                        && topicStats.subscriptions.get(subscriptionName).consumers.isEmpty();
+            } catch (PulsarAdminException e) {
+                return false;
+            }
+        }, 20, 150);
+
+        //change the schema ,the function should not run, resulting in no messages to consume
+        schemaInput.put(sourceTopic, "{\"schemaType\":\"AVRO\",\"schemaProperties\":{\"__jsr310ConversionEnabled\":\"false\",\"__alwaysAllowNull\":\"false\"}}");
+        localRunner = LocalRunner.builder()
+                .functionConfig(functionConfig)
+                .clientAuthPlugin(AuthenticationTls.class.getName())
+                .clientAuthParams(String.format("tlsCertFile:%s,tlsKeyFile:%s", TLS_CLIENT_CERT_FILE_PATH, TLS_CLIENT_KEY_FILE_PATH))
+                .useTls(true)
+                .tlsTrustCertFilePath(TLS_TRUST_CERT_FILE_PATH)
+                .tlsAllowInsecureConnection(true)
+                .tlsHostNameVerificationEnabled(false)
+                .brokerServiceUrl(pulsar.getBrokerServiceUrlTls()).build();
+        localRunner.start(false);
+
+        producer.newMessage().property(propertyKey, propertyValue).value(new AvroTestObject()).send();
+        Message<GenericRecord> msg = consumer.receive(2, TimeUnit.SECONDS);
+        assertEquals(msg, null);
+
+        producer.close();
+        consumer.close();
+        localRunner.stop();
+    }
+
     @Test(timeOut = 20000)
     public void testE2EPulsarFunctionLocalRun() throws Exception {
         testE2EPulsarFunctionLocalRun(null);
     }
 
+    @Test(timeOut = 30000)
+    public void testAvroFunctionLocalRun() throws Exception {
+        testAvroFunctionLocalRun(null);
+    }
+
     @Test(timeOut = 20000)
     public void testE2EPulsarFunctionLocalRunWithJar() throws Exception {
         String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile();

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java

@@ -231,8 +231,10 @@ abstract class FunctionDetailsCommand extends BaseCommand {
         protected String DEPRECATED_customSerdeInputString;
         @Parameter(names = "--custom-serde-inputs", description = "The map of input topics to SerDe class names (as a JSON string)")
         protected String customSerdeInputString;
-        @Parameter(names = "--custom-schema-inputs", description = "The map of input topics to Schema class names (as a JSON string)")
+        @Parameter(names = "--custom-schema-inputs", description = "The map of input topics to Schema properties (as a JSON string)")
         protected String customSchemaInputString;
+        @Parameter(names = "--custom-schema-outputs", description = "The map of input topics to Schema properties (as a JSON string)")
+        protected String customSchemaOutputString;
         // for backwards compatibility purposes
         @Parameter(names = "--outputSerdeClassName", description = "The SerDe class to be used for messages output by the function", hidden = true)
         protected String DEPRECATED_outputSerdeClassName;
@@ -368,6 +370,11 @@ void processArguments() throws Exception {
                 Map<String, String> customschemaInputMap = new Gson().fromJson(customSchemaInputString, type);
                 functionConfig.setCustomSchemaInputs(customschemaInputMap);
             }
+            if (null != customSchemaOutputString) {
+                Type type = new TypeToken<Map<String, String>>() {}.getType();
+                Map<String, String> customSchemaOutputMap = new Gson().fromJson(customSchemaOutputString, type);
+                functionConfig.setCustomSchemaOutputs(customSchemaOutputMap);
+            }
             if (null != topicsPattern) {
                 functionConfig.setTopicsPattern(topicsPattern);
             }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/SchemaDefinitionBuilderImpl.java

@@ -117,6 +117,12 @@ public SchemaDefinitionBuilder<T> withSupportSchemaVersioning(boolean supportSch
     @Override
     public SchemaDefinitionBuilder<T> withProperties(Map<String,String> properties) {
         this.properties = properties;
+        if (properties.containsKey(ALWAYS_ALLOW_NULL)) {
+            alwaysAllowNull = Boolean.parseBoolean(properties.get(ALWAYS_ALLOW_NULL));
+        }
+        if (properties.containsKey(ALWAYS_ALLOW_NULL)) {
+            jsr310ConversionEnabled = Boolean.parseBoolean(properties.get(JSR310_CONVERSION_ENABLED));
+        }
         return this;
     }
 

pulsar-common/src/main/java/org/apache/pulsar/common/functions/ConsumerConfig.java

@@ -18,12 +18,16 @@
  */
 package org.apache.pulsar.common.functions;
 
+import java.util.HashMap;
+import java.util.Map;
+
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.NoArgsConstructor;
 
+
 /**
  * Configuration of a consumer.
  */
@@ -36,5 +40,12 @@ public class ConsumerConfig {
     private String schemaType;
     private String serdeClassName;
     private boolean isRegexPattern;
+    @Builder.Default
+    private Map<String, String> schemaProperties = new HashMap<>();
     private Integer receiverQueueSize;
+
+    public ConsumerConfig(String schemaType) {
+        this.schemaType = schemaType;
+    }
+
 }

pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java

@@ -68,6 +68,7 @@ public enum Runtime {
     private Map<String, String> customSerdeInputs;
     private String topicsPattern;
     private Map<String, String> customSchemaInputs;
+    private Map<String, String> customSchemaOutputs;
 
     /**
      * A generalized way of specifying inputs.

pulsar-common/src/main/java/org/apache/pulsar/common/io/SinkConfig.java

@@ -54,6 +54,8 @@ public class SinkConfig {
 
     private Map<String, String> topicToSchemaType;
 
+    private Map<String, String> topicToSchemaProperties;
+
     private Map<String, ConsumerConfig> inputSpecs;
 
     private Integer maxMessageRetries;

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java

@@ -682,6 +682,7 @@ private void setupInput(ContextImpl contextImpl) throws Exception {
                 } else if (conf.getSerdeClassName() != null && !conf.getSerdeClassName().isEmpty()) {
                     consumerConfig.setSerdeClassName(conf.getSerdeClassName());
                 }
+                consumerConfig.setSchemaProperties(conf.getSchemaPropertiesMap());
                 if (conf.hasReceiverQueueSize()) {
                     consumerConfig.setReceiverQueueSize(conf.getReceiverQueueSize().getValue());
                 }
@@ -796,6 +797,7 @@ private void setupOutput(ContextImpl contextImpl) throws Exception {
                 }
 
                 pulsarSinkConfig.setTypeClassName(sinkSpec.getTypeClassName());
+                pulsarSinkConfig.setSchemaProperties(sinkSpec.getSchemaPropertiesMap());
 
                 object = new PulsarSink(this.client, pulsarSinkConfig, this.properties, this.stats, this.functionClassLoader);
             }

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java

@@ -32,6 +32,7 @@
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.TypedMessageBuilder;
 import org.apache.pulsar.client.impl.schema.KeyValueSchema;
+import org.apache.pulsar.common.functions.ConsumerConfig;
 import org.apache.pulsar.common.functions.FunctionConfig;
 import org.apache.pulsar.common.schema.KeyValueEncodingType;
 import org.apache.pulsar.functions.api.Record;
@@ -339,13 +340,16 @@ Schema<T> initializeSchema() throws ClassNotFoundException {
             // return type is 'void', so there's no schema to check
             return null;
         }
-
+        ConsumerConfig consumerConfig = new ConsumerConfig();
+        consumerConfig.setSchemaProperties(pulsarSinkConfig.getSchemaProperties());
         if (!StringUtils.isEmpty(pulsarSinkConfig.getSchemaType())) {
+            consumerConfig.setSchemaType(pulsarSinkConfig.getSchemaType());
             return (Schema<T>) topicSchema.getSchema(pulsarSinkConfig.getTopic(), typeArg,
-                    pulsarSinkConfig.getSchemaType(), false);
+                    consumerConfig, false);
         } else {
+            consumerConfig.setSchemaType(pulsarSinkConfig.getSerdeClassName());
             return (Schema<T>) topicSchema.getSchema(pulsarSinkConfig.getTopic(), typeArg,
-                    pulsarSinkConfig.getSerdeClassName(), false, functionClassLoader);
+                    consumerConfig, false, functionClassLoader);
         }
     }
 }

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSinkConfig.java

@@ -23,6 +23,8 @@
 import lombok.ToString;
 import org.apache.pulsar.common.functions.FunctionConfig;
 
+import java.util.Map;
+
 @Getter
 @Setter
 @ToString
@@ -31,6 +33,7 @@ public class PulsarSinkConfig {
     private String topic;
     private String serdeClassName;
     private String schemaType;
+    private Map<String, String> schemaProperties;
     private String typeClassName;
     private boolean forwardSourceMessageProperty;
 }

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java

@@ -166,7 +166,7 @@ Map<String, ConsumerConfig<T>> setupConsumerConfigs() throws ClassNotFoundExcept
             if (conf.getSerdeClassName() != null && !conf.getSerdeClassName().isEmpty()) {
                 schema = (Schema<T>) topicSchema.getSchema(topic, typeArg, conf.getSerdeClassName(), true);
             } else {
-                schema = (Schema<T>) topicSchema.getSchema(topic, typeArg, conf.getSchemaType(), true);
+                schema = (Schema<T>) topicSchema.getSchema(topic, typeArg, conf, true);
             }
             configs.put(topic,
                     ConsumerConfig.<T> builder().schema(schema).isRegexPattern(conf.isRegexPattern()).receiverQueueSize(conf.getReceiverQueueSize()).build());