This is an automated email from the ASF dual-hosted git repository. jerrypeng 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 a8cd98c Fix update cli source sink (#4061) a8cd98c is described below commit a8cd98cd87dde7fa17ea38ff212eebd14fc95835 Author: Boyang Jerry Peng <jerry.boyang.p...@gmail.com> AuthorDate: Wed Apr 17 17:42:38 2019 -0700 Fix update cli source sink (#4061) * fix bug in source and sink cli update * fix import * fix logic * fix tests --- .../java/org/apache/pulsar/admin/cli/CmdSinks.java | 7 +++- .../org/apache/pulsar/admin/cli/CmdSources.java | 8 ++++- .../org/apache/pulsar/admin/cli/TestCmdSinks.java | 39 ++++++++++++++++++++-- .../apache/pulsar/admin/cli/TestCmdSources.java | 36 ++++++++++++++++++++ .../pulsar/common/functions/FunctionConfig.java | 2 +- .../org/apache/pulsar/common/io/SinkConfig.java | 2 +- .../functions/utils/FunctionConfigUtils.java | 9 +++++ .../pulsar/functions/utils/SinkConfigUtils.java | 9 +++++ 8 files changed, 105 insertions(+), 7 deletions(-) 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 711ed88..18ffb53 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 @@ -222,7 +222,12 @@ public class CmdSinks extends CmdBase { } protected void validateSinkConfigs(SinkConfig sinkConfig) { - org.apache.pulsar.common.functions.Utils.inferMissingArguments(sinkConfig); + if (sinkConfig.getTenant() == null) { + sinkConfig.setTenant(PUBLIC_TENANT); + } + if (sinkConfig.getNamespace() == null) { + sinkConfig.setNamespace(DEFAULT_NAMESPACE); + } } } 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 a200c5e..07071a7 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 @@ -215,6 +215,7 @@ public class CmdSources extends CmdBase { @Parameters(commandDescription = "Update a Pulsar IO source connector") protected class UpdateSource extends SourceDetailsCommand { + @Override void runCmd() throws Exception { if (Utils.isFunctionPackageUrlSupported(sourceConfig.getArchive())) { @@ -226,7 +227,12 @@ public class CmdSources extends CmdBase { } protected void validateSourceConfigs(SourceConfig sourceConfig) { - org.apache.pulsar.common.functions.Utils.inferMissingArguments(sourceConfig); + if (sourceConfig.getTenant() == null) { + sourceConfig.setTenant(PUBLIC_TENANT); + } + if (sourceConfig.getNamespace() == null) { + sourceConfig.setNamespace(DEFAULT_NAMESPACE); + } } } diff --git a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSinks.java b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSinks.java index 5d8a303..98d31a4 100644 --- a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSinks.java +++ b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSinks.java @@ -165,7 +165,6 @@ public class TestCmdSinks { @Test public void testMissingInput() throws Exception { SinkConfig sinkConfig = getSinkConfig(); - sinkConfig.setInputSpecs(new HashMap<>()); sinkConfig.setInputs(null); testCmdSinkCliMissingArgs( TENANT, @@ -190,7 +189,6 @@ public class TestCmdSinks { SinkConfig sinkConfig = getSinkConfig(); sinkConfig.setTopicToSerdeClassName(null); sinkConfig.setTopicToSchemaType(null); - sinkConfig.setInputSpecs(new HashMap<>()); testCmdSinkCliMissingArgs( TENANT, NAMESPACE, @@ -212,7 +210,6 @@ public class TestCmdSinks { @Test public void testMissingTopicPattern() throws Exception { SinkConfig sinkConfig = getSinkConfig(); - sinkConfig.getInputSpecs().clear(); sinkConfig.setTopicsPattern(null); testCmdSinkCliMissingArgs( TENANT, @@ -677,4 +674,40 @@ public class TestCmdSinks { verify(sink).deleteSink(eq(TENANT), eq(NAMESPACE), null); } + + @Test + public void testUpdateSink() throws Exception { + + updateSink.name = "my-sink"; + + updateSink.archive = "new-archive"; + + updateSink.processArguments(); + + updateSink.runCmd(); + + verify(sink).updateSink(eq(SinkConfig.builder() + .tenant(PUBLIC_TENANT) + .namespace(DEFAULT_NAMESPACE) + .name(updateSink.name) + .archive(updateSink.archive) + .build()), eq(updateSink.archive)); + + + updateSink.archive = null; + + updateSink.parallelism = 2; + + updateSink.processArguments(); + + updateSink.runCmd(); + + verify(sink).updateSink(eq(SinkConfig.builder() + .tenant(PUBLIC_TENANT) + .namespace(DEFAULT_NAMESPACE) + .name(updateSink.name) + .parallelism(2) + .build()), eq(null)); + + } } diff --git a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSources.java b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSources.java index ac2dab9..04fbf15 100644 --- a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSources.java +++ b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSources.java @@ -569,4 +569,40 @@ public class TestCmdSources { verify(source).deleteSource(eq(TENANT), eq(NAMESPACE), null); } + + @Test + public void testUpdateSource() throws Exception { + + updateSource.name = "my-source"; + + updateSource.archive = "new-archive"; + + updateSource.processArguments(); + + updateSource.runCmd(); + + verify(source).updateSource(eq(SourceConfig.builder() + .tenant(PUBLIC_TENANT) + .namespace(DEFAULT_NAMESPACE) + .name(updateSource.name) + .archive(updateSource.archive) + .build()), eq(updateSource.archive)); + + + updateSource.archive = null; + + updateSource.parallelism = 2; + + updateSource.processArguments(); + + updateSource.runCmd(); + + verify(source).updateSource(eq(SourceConfig.builder() + .tenant(PUBLIC_TENANT) + .namespace(DEFAULT_NAMESPACE) + .name(updateSource.name) + .parallelism(2) + .build()), eq(null)); + + } } \ No newline at end of file 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 index dc9023a..2b8883d 100644 --- 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 @@ -64,7 +64,7 @@ public class FunctionConfig { /** * A generalized way of specifying inputs */ - private Map<String, ConsumerConfig> inputSpecs = new TreeMap<>(); + private Map<String, ConsumerConfig> inputSpecs; private String output; 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 index d6dd92c..6e51cbb 100644 --- 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 @@ -51,7 +51,7 @@ public class SinkConfig { private Map<String, String> topicToSchemaType; - private Map<String, ConsumerConfig> inputSpecs = new TreeMap<>(); + private Map<String, ConsumerConfig> inputSpecs; private Map<String, Object> configs; // This is a map of secretName(aka how the secret is going to be diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java index 6f9ab27..e245216 100644 --- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java +++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java @@ -607,6 +607,15 @@ public class FunctionConfigUtils { if (!StringUtils.isEmpty(newConfig.getClassName())) { mergedConfig.setClassName(newConfig.getClassName()); } + + if (newConfig.getInputSpecs() == null) { + newConfig.setInputSpecs(new HashMap<>()); + } + + if (mergedConfig.getInputSpecs() == null) { + mergedConfig.setInputSpecs(new HashMap<>()); + } + if (newConfig.getInputs() != null) { newConfig.getInputs().forEach((topicName -> { newConfig.getInputSpecs().put(topicName, 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 index d0e71c8..9159f4f 100644 --- 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 @@ -422,6 +422,15 @@ public class SinkConfigUtils { if (!StringUtils.isEmpty(newConfig.getSourceSubscriptionName()) && !newConfig.getSourceSubscriptionName().equals(existingConfig.getSourceSubscriptionName())) { throw new IllegalArgumentException("Subscription Name cannot be altered"); } + + if (newConfig.getInputSpecs() == null) { + newConfig.setInputSpecs(new HashMap<>()); + } + + if (mergedConfig.getInputSpecs() == null) { + mergedConfig.setInputSpecs(new HashMap<>()); + } + if (newConfig.getInputs() != null) { newConfig.getInputs().forEach((topicName -> { newConfig.getInputSpecs().put(topicName,