SAMZA-1806: Allow `task.broadcast.inputs` to be set to empty string
authorAbhishek Shivanna <abhisheks91@gmail.com>
Fri, 10 Aug 2018 22:55:36 +0000 (15:55 -0700)
committerJagadish <jvenkatraman@linkedin.com>
Fri, 10 Aug 2018 22:55:36 +0000 (15:55 -0700)
This fix addresses the issue that getBroadcastSystemStreamPartitions
in TaskConfigJava throws an IllegalArgumentException exception when
`task.broadcast.inputs` is set to an empty string.

Author: Abhishek Shivanna <abhisheks91@gmail.com>

Reviewers: Bharath K <bkumarasubramanian@linkedin.com>

Closes #604 from abhishekshivanna/master

samza-api/src/main/java/org/apache/samza/config/Config.java
samza-api/src/test/java/org/apache/samza/config/TestConfig.java

index 7abe557..d4164c6 100644 (file)
@@ -154,6 +154,9 @@ public abstract class Config implements Map<String, String> {
       return defaultValue;
 
     String value = get(k);
+    if (value.trim().isEmpty()) {
+      return defaultValue;
+    }
     String[] pieces = value.split("\\s*,\\s*");
     return Arrays.asList(pieces);
   }
index 7d9d56c..c3ca145 100644 (file)
@@ -19,6 +19,8 @@
 
 package org.apache.samza.config;
 
+import java.util.Collections;
+import java.util.List;
 import org.junit.Test;
 
 import java.util.HashMap;
@@ -85,4 +87,43 @@ public class TestConfig {
     assertEquals(Config.SENSITIVE_MASK, sanitized.get("sensitive.key3"));
     assertEquals(Config.SENSITIVE_MASK, sanitized.get("sensitive.key4"));
   }
+
+  @Test
+  public void testGetList() {
+    Map<String, String> m = new HashMap<String, String>() {
+      {
+        put("key1", " ");
+        put("key2", "");
+        put("key3", "  value1  ");
+        put("key4", "value1,value2");
+        put("key5", "value1, value2");
+        put("key6", "value1  ,   value2");
+      }
+    };
+
+    Config config = new MapConfig(m);
+    List<String> list = config.getList("key1", Collections.<String>emptyList());
+    assertEquals(0, list.size());
+
+    list = config.getList("key2", Collections.<String>emptyList());
+    assertEquals(0, list.size());
+
+    list = config.getList("key3");
+    assertEquals("  value1  ", list.get(0));
+
+    list = config.getList("key4");
+    assertEquals("value1", list.get(0));
+    assertEquals("value2", list.get(1));
+
+    list = config.getList("key5");
+    assertEquals("value1", list.get(0));
+    assertEquals("value2", list.get(1));
+
+    list = config.getList("key6");
+    assertEquals("value1", list.get(0));
+    assertEquals("value2", list.get(1));
+
+    list = config.getList("UndefinedKey", Collections.<String>emptyList());
+    assertEquals(0, list.size());
+  }
 }