Add ability to specify runtime flags in functions/sources/sinks (apac…

…he#3872)
willome · Mar 29, 2019 · 0e378dc · 0e378dc
1 parent fa3bc7e
commit 0e378dc
Show file tree

Hide file tree

Showing 11 changed files with 43 additions and 0 deletions.
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java
@@ -48,6 +48,10 @@ public enum Runtime {
         PYTHON
     }
 
+    // Any flags that you want to pass to the runtime.
+    // note that in thread mode, these flags will have no impact
+    private String runtimeFlags;
+
     private String tenant;
     private String namespace;
     private String name;

diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/io/SinkConfig.java b/pulsar-common/src/main/java/org/apache/pulsar/common/io/SinkConfig.java
@@ -69,4 +69,7 @@ public class SinkConfig {
     private String archive;
     // Whether the subscriptions the functions created/used should be deleted when the functions is deleted
     private Boolean cleanupSubscription;
+
+    // Any flags that you want to pass to the runtime.
+    private String runtimeFlags;
 }
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java b/pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java
@@ -56,4 +56,6 @@ public class SourceConfig {
     private Resources resources;
 
     private String archive;
+    // Any flags that you want to pass to the runtime.
+    private String runtimeFlags;
 }
diff --git a/pulsar-functions/proto/src/main/proto/Function.proto b/pulsar-functions/proto/src/main/proto/Function.proto
@@ -66,6 +66,7 @@ message FunctionDetails {
     Resources resources = 13;
     string packageUrl = 14; //present only if function submitted with package-url
     RetryDetails retryDetails = 15;
+    string runtimeFlags = 17;
 }
 
 message ConsumerSpec {

diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java
@@ -140,6 +140,9 @@ public static List<String> getCmd(InstanceConfig instanceConfig,
                     "%s-%s",
                     instanceConfig.getFunctionDetails().getName(),
                     shardId));
+            if (!isEmpty(instanceConfig.getFunctionDetails().getRuntimeFlags())) {
+                args.add(instanceConfig.getFunctionDetails().getRuntimeFlags());
+            }
             if (instanceConfig.getFunctionDetails().getResources() != null) {
                 Function.Resources resources = instanceConfig.getFunctionDetails().getResources();
                 if (resources.getRam() != 0) {
@@ -151,6 +154,9 @@ public static List<String> getCmd(InstanceConfig instanceConfig,
             args.add(originalCodeFileName);
         } else if (instanceConfig.getFunctionDetails().getRuntime() == Function.FunctionDetails.Runtime.PYTHON) {
             args.add("python");
+            if (!isEmpty(instanceConfig.getFunctionDetails().getRuntimeFlags())) {
+                args.add(instanceConfig.getFunctionDetails().getRuntimeFlags());
+            }
             args.add(instanceFile);
             args.add("--py");
             args.add(originalCodeFileName);

diff --git a/...-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java b/...-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java
@@ -217,6 +217,10 @@ public static FunctionDetails convert(FunctionConfig functionConfig, ClassLoader
         bldr.setDisk(resources.getDisk());
         functionDetailsBuilder.setResources(bldr);
 
+        if (!StringUtils.isEmpty(functionConfig.getRuntimeFlags())) {
+            functionDetailsBuilder.setRuntimeFlags(functionConfig.getRuntimeFlags());
+        }
+
         return functionDetailsBuilder.build();
     }
 
@@ -308,6 +312,10 @@ public static FunctionConfig convertFromDetails(FunctionDetails functionDetails)
             functionConfig.setResources(resources);
         }
 
+        if (!isEmpty(functionDetails.getRuntimeFlags())) {
+            functionConfig.setRuntimeFlags(functionDetails.getRuntimeFlags());
+        }
+
         return functionConfig;
     }
 

diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java
@@ -190,6 +190,10 @@ public static FunctionDetails convert(SinkConfig sinkConfig, ExtractedSinkDetail
         bldr.setDisk(resources.getDisk());
         functionDetailsBuilder.setResources(bldr);
 
+        if (isNotBlank(sinkConfig.getRuntimeFlags())) {
+            functionDetailsBuilder.setRuntimeFlags(sinkConfig.getRuntimeFlags());
+        }
+
         return functionDetailsBuilder.build();
     }
 
@@ -249,6 +253,10 @@ public static SinkConfig convertFromDetails(FunctionDetails functionDetails) {
             resources.setDisk(functionDetails.getResources().getDisk());
         }
 
+        if (isNotBlank(functionDetails.getRuntimeFlags())) {
+            sinkConfig.setRuntimeFlags(functionDetails.getRuntimeFlags());
+        }
+
         return sinkConfig;
     }
 

diff --git a/...ar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java b/...ar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java
@@ -133,6 +133,10 @@ public static FunctionDetails convert(SourceConfig sourceConfig, ExtractedSource
         bldr.setDisk(resources.getDisk());
         functionDetailsBuilder.setResources(bldr);
 
+        if (!org.apache.commons.lang3.StringUtils.isEmpty(sourceConfig.getRuntimeFlags())) {
+            functionDetailsBuilder.setRuntimeFlags(sourceConfig.getRuntimeFlags());
+        }
+
         return functionDetailsBuilder.build();
     }
 
@@ -174,6 +178,10 @@ public static SourceConfig convertFromDetails(FunctionDetails functionDetails) {
             resources.setDisk(functionDetails.getResources().getDisk());
             sourceConfig.setResources(resources);
         }
+
+        if (!org.apache.commons.lang3.StringUtils.isEmpty(functionDetails.getRuntimeFlags())) {
+            sourceConfig .setRuntimeFlags(functionDetails.getRuntimeFlags());
+        }
         return sourceConfig;
     }
 

diff --git a/...ctions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java b/...ctions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java
@@ -62,6 +62,7 @@ public void testConvertBackFidelity() {
         functionConfig.setUserConfig(new HashMap<>());
         functionConfig.setAutoAck(true);
         functionConfig.setTimeoutMs(2000l);
+        functionConfig.setRuntimeFlags("-DKerberos");
         Function.FunctionDetails functionDetails = FunctionConfigUtils.convert(functionConfig, null);
         FunctionConfig convertedConfig = FunctionConfigUtils.convertFromDetails(functionDetails);
 

diff --git a/...-functions/utils/src/test/java/org/apache/pulsar/functions/utils/SinkConfigUtilsTest.java b/...-functions/utils/src/test/java/org/apache/pulsar/functions/utils/SinkConfigUtilsTest.java
@@ -57,6 +57,7 @@ public void testConvertBackFidelity() throws IOException  {
         sinkConfig.setRetainOrdering(false);
         sinkConfig.setAutoAck(true);
         sinkConfig.setTimeoutMs(2000l);
+        sinkConfig.setRuntimeFlags("-DKerberos");
         Function.FunctionDetails functionDetails = SinkConfigUtils.convert(sinkConfig, new SinkConfigUtils.ExtractedSinkDetails(null, null));
         SinkConfig convertedConfig = SinkConfigUtils.convertFromDetails(functionDetails);
         assertEquals(

diff --git a/...unctions/utils/src/test/java/org/apache/pulsar/functions/utils/SourceConfigUtilsTest.java b/...unctions/utils/src/test/java/org/apache/pulsar/functions/utils/SourceConfigUtilsTest.java
@@ -49,6 +49,7 @@ public void testConvertBackFidelity() throws IOException  {
         sourceConfig.setTopicName("test-output");
         sourceConfig.setSerdeClassName("test-serde");
         sourceConfig.setParallelism(1);
+        sourceConfig.setRuntimeFlags("-DKerberos");
         sourceConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE);
         sourceConfig.setConfigs(new HashMap<>());
         Function.FunctionDetails functionDetails = SourceConfigUtils.convert(sourceConfig, new SourceConfigUtils.ExtractedSourceDetails(null, null));