[pulsar] branch master updated: Infer paralellism as well (#2999)

[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 2d1db63  Infer paralellism as well (#2999)
2d1db63 is described below

commit 2d1db63eb2b7498dbc74e0704bde5ac13dd99655
Author: Sanjeev Kulkarni <sanjee...@gmail.com>
AuthorDate: Sat Nov 17 09:30:11 2018 -0800

    Infer paralellism as well (#2999)
    
    * Infer paralellism as well
    
    * Fixed unittests
---
 .../org/apache/pulsar/admin/cli/TestCmdSinks.java  | 22 ----------------------
 .../apache/pulsar/admin/cli/TestCmdSources.java    | 19 -------------------
 .../org/apache/pulsar/common/functions/Utils.java  |  6 ++++++
 3 files changed, 6 insertions(+), 41 deletions(-)

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 d1fad7a..a52b7b8 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
@@ -254,28 +254,6 @@ public class TestCmdSinks {
                 sinkConfig
         );
     }
-
-    @Test
-    public void testMissingParallelism() throws Exception {
-        SinkConfig sinkConfig = getSinkConfig();
-        sinkConfig.setParallelism(null);
-        testCmdSinkCliMissingArgs(
-                TENANT,
-                NAMESPACE,
-                NAME,
-                INPUTS,
-                TOPIC_PATTERN,
-                CUSTOM_SERDE_INPUT_STRING,
-                PROCESSING_GUARANTEES,
-                null,
-                JAR_FILE_PATH,
-                CPU,
-                RAM,
-                DISK,
-                SINK_CONFIG_STRING,
-                sinkConfig
-        );
-    }
     
     @Test(expectedExceptions = ParameterException.class, 
expectedExceptionsMessageRegExp = "Sink archive not specfied")
     public void testMissingArchive() throws Exception {
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 6a64acf..2f63626 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
@@ -178,25 +178,6 @@ public class TestCmdSources {
         );
     }
 
-    @Test
-    public void testMissingParallelism() throws Exception {
-        SourceConfig sourceConfig = getSourceConfig();
-        sourceConfig.setParallelism(null);
-        testCmdSourceCliMissingArgs(
-                TENANT,
-                NAMESPACE,
-                NAME,
-                TOPIC_NAME, SERDE_CLASS_NAME, PROCESSING_GUARANTEES,
-                null,
-                JAR_FILE_PATH,
-                CPU,
-                RAM,
-                DISK,
-                SINK_CONFIG_STRING,
-                sourceConfig
-        );
-    }
-
     @Test(expectedExceptions = ParameterException.class, 
expectedExceptionsMessageRegExp = "Source archive not specfied")
     public void testMissingArchive() throws Exception {
         SourceConfig sourceConfig = getSourceConfig();
diff --git 
a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/Utils.java 
b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/Utils.java
index 8992cd0..b226c90 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/Utils.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/Utils.java
@@ -59,6 +59,9 @@ public class Utils {
         if (sourceConfig.getNamespace() == null) {
             sourceConfig.setNamespace(DEFAULT_NAMESPACE);
         }
+        if (sourceConfig.getParallelism() == null) {
+            sourceConfig.setParallelism(1);
+        }
     }
 
     public static void inferMissingArguments(SinkConfig sinkConfig) {
@@ -68,5 +71,8 @@ public class Utils {
         if (sinkConfig.getNamespace() == null) {
             sinkConfig.setNamespace(DEFAULT_NAMESPACE);
         }
+        if (sinkConfig.getParallelism() == null) {
+            sinkConfig.setParallelism(1);
+        }
     }
 }