[incubator-pulsar] branch master updated: Have the ability to provide source/sink config in cmdline for sources/sinks (#1757)

[mm...@apache.org]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new 8b2dfd0  Have the ability to provide source/sink config in cmdline for sources/sinks (#1757)
8b2dfd0 is described below

commit 8b2dfd04d8bac20ade9438c953dd74337835098f
Author: Sanjeev Kulkarni <sa...@gmail.com>
AuthorDate: Wed May 9 19:28:13 2018 -0700

    Have the ability to provide source/sink config in cmdline for sources/sinks (#1757)
---
 .../src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java       | 8 ++++++++
 .../src/main/java/org/apache/pulsar/admin/cli/CmdSources.java     | 8 ++++++++
 .../main/java/org/apache/pulsar/functions/utils/SinkConfig.java   | 2 +-
 3 files changed, 17 insertions(+), 1 deletion(-)

diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
index a6319c1..36fdf46 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
@@ -133,6 +133,8 @@ public class CmdSinks extends CmdBase {
         @Parameter(names = "--sinkConfigFile", description = "The path to a YAML config file specifying the "
                 + "sink's configuration")
         protected String sinkConfigFile;
+        @Parameter(names = "--sinkConfig", description = "Sink config key/values")
+        protected String sinkConfigString;
 
         protected SinkConfig sinkConfig;
 
@@ -199,6 +201,12 @@ public class CmdSinks extends CmdBase {
             if (null == jarFile) {
                 throw new IllegalArgumentException("Connector JAR not specfied");
             }
+
+            if (null != sinkConfigString) {
+                Type type = new TypeToken<Map<String, String>>(){}.getType();
+                Map<String, Object> sinkConfigMap = new Gson().fromJson(sinkConfigString, type);
+                sinkConfig.setConfigs(sinkConfigMap);
+            }
         }
 
         @Override
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
index 7ff908a..79dbb50 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
@@ -133,6 +133,8 @@ public class CmdSources extends CmdBase {
         @Parameter(names = "--sourceConfigFile", description = "The path to a YAML config file specifying the "
                 + "source's configuration")
         protected String sourceConfigFile;
+        @Parameter(names = "--sourceConfig", description = "Source config key/values")
+        protected String sourceConfigString;
 
         protected SourceConfig sourceConfig;
 
@@ -184,6 +186,12 @@ public class CmdSources extends CmdBase {
             if (null == jarFile) {
                 throw new IllegalArgumentException("Connector JAR not specfied");
             }
+
+            if (null != sourceConfigString) {
+                Type type = new TypeToken<Map<String, String>>(){}.getType();
+                Map<String, Object> sourceConfigMap = new Gson().fromJson(sourceConfigString, type);
+                sourceConfig.setConfigs(sourceConfigMap);
+            }
         }
 
         @Override
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfig.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfig.java
index 332ee4c..bfc9a43 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfig.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfig.java
@@ -38,7 +38,7 @@ public class SinkConfig {
     private String name;
     private String className;
     private Map<String, String> topicToSerdeClassName;
-    private Map<String, String> configs = new HashMap<>();
+    private Map<String, Object> configs = new HashMap<>();
     private int parallelism = 1;
     private FunctionConfig.ProcessingGuarantees processingGuarantees;
 }

-- 
To stop receiving notification emails like this one, please contact
mmerli@apache.org.