[pulsar] branch master updated: Infer missing arguments before starting localrun (#2937)

[sanjeevrk]

This is an automated email from the ASF dual-hosted git repository.

sanjeevrk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new d1c531a  Infer missing arguments before starting localrun (#2937)
d1c531a is described below

commit d1c531acdcbe0bc01f470ff67b6c1c41f767adfa
Author: Sanjeev Kulkarni <sanjee...@gmail.com>
AuthorDate: Mon Nov 5 21:28:12 2018 -0800

    Infer missing arguments before starting localrun (#2937)
---
 .../main/java/org/apache/pulsar/functions/runtime/LocalRunner.java    | 4 ++++
 1 file changed, 4 insertions(+)

diff --git 
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/LocalRunner.java
 
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/LocalRunner.java
index e85a45d..958dbcf 100644
--- 
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/LocalRunner.java
+++ 
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/LocalRunner.java
@@ -47,6 +47,7 @@ import org.apache.pulsar.functions.utils.*;
 import org.apache.pulsar.functions.utils.io.ConnectorUtils;
 import org.apache.pulsar.functions.utils.io.Connectors;
 
+import static org.apache.pulsar.common.functions.Utils.inferMissingArguments;
 import static org.apache.pulsar.functions.utils.Utils.*;
 
 @Slf4j
@@ -94,6 +95,7 @@ public class LocalRunner {
         int parallelism;
         if (!StringUtils.isEmpty(functionConfigString)) {
             FunctionConfig functionConfig = new 
Gson().fromJson(functionConfigString, FunctionConfig.class);
+            FunctionConfigUtils.inferMissingArguments(functionConfig);
             ClassLoader classLoader = null;
             parallelism = functionConfig.getParallelism();
             if (functionConfig.getRuntime() == FunctionConfig.Runtime.JAVA) {
@@ -113,6 +115,7 @@ public class LocalRunner {
             functionDetails = FunctionConfigUtils.convert(functionConfig, 
classLoader);
         } else if (!StringUtils.isEmpty(sourceConfigString)) {
             SourceConfig sourceConfig = new 
Gson().fromJson(sourceConfigString, SourceConfig.class);
+            inferMissingArguments(sourceConfig);
             String builtInSource = isBuiltInSource(sourceConfig.getArchive());
             if (builtInSource != null) {
                 sourceConfig.setArchive(builtInSource);
@@ -132,6 +135,7 @@ public class LocalRunner {
             functionDetails = SourceConfigUtils.convert(sourceConfig, 
classLoader);
         } else {
             SinkConfig sinkConfig = new Gson().fromJson(sinkConfigString, 
SinkConfig.class);
+            inferMissingArguments(sinkConfig);
             String builtInSink = isBuiltInSource(sinkConfig.getArchive());
             if (builtInSink != null) {
                 sinkConfig.setArchive(builtInSink);