Source - add default values and Fix logic for prefix/whitelist/blacklist

Fixes shikhar#4

Author: shikhar

README.rst

@@ -186,20 +186,24 @@ Configuration options
   * Default: "${table}"
   * Importance: high
 
+``tables.prefix``
+  Prefix for DynamoDB tables to source from.
+
+  * Type: string
+  * Default: ""
+  * Importance: medium
+
 ``tables.whitelist``
   Whitelist for DynamoDB tables to source from.
 
   * Type: list
+  * Default: ""
   * Importance: medium
 
 ``tables.blacklist``
   Blacklist for DynamoDB tables to source from.
 
   * Type: list
+  * Default: ""
   * Importance: medium
 
-``tables.prefix``
-  Prefix for DynamoDB tables to source from.
-
-  * Type: string
-  * Importance: medium

src/main/java/dynamok/source/ConnectorConfig.java

@@ -22,6 +22,7 @@
 import org.apache.kafka.common.config.ConfigException;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
@@ -42,11 +43,11 @@ private enum Keys {
                     throw new ConfigException("Invalid AWS region: " + regionName);
                 }
             }, ConfigDef.Importance.HIGH, "AWS region for the source DynamoDB.")
-            .define(Keys.TABLES_PREFIX, ConfigDef.Type.STRING, null,
+            .define(Keys.TABLES_PREFIX, ConfigDef.Type.STRING, "",
                     ConfigDef.Importance.MEDIUM, "Prefix for DynamoDB tables to source from.")
-            .define(Keys.TABLES_WHITELIST, ConfigDef.Type.LIST, null,
+            .define(Keys.TABLES_WHITELIST, ConfigDef.Type.LIST, Collections.emptyList(),
                     ConfigDef.Importance.MEDIUM, "Whitelist for DynamoDB tables to source from.")
-            .define(Keys.TABLES_BLACKLIST, ConfigDef.Type.LIST, null,
+            .define(Keys.TABLES_BLACKLIST, ConfigDef.Type.LIST, Collections.emptyList(),
                     ConfigDef.Importance.MEDIUM, "Blacklist for DynamoDB tables to source from.")
             .define(Keys.TOPIC_FORMAT, ConfigDef.Type.STRING, "${table}",
                     ConfigDef.Importance.HIGH, "Format string for destination Kafka topic, use ``${table}`` as placeholder for source table name.");

src/main/java/dynamok/source/DynamoDbSourceConnector.java

@@ -88,9 +88,7 @@ public void start(Map<String, String> props) {
             final ListTablesResult listResult = client.listTables(lastEvaluatedTableName);
 
             for (String tableName : listResult.getTableNames()) {
-                if ((config.tablesPrefix == null || tableName.startsWith(config.tablesPrefix)) &&
-                        (config.tablesBlacklist == null || !config.tablesBlacklist.contains(tableName)) &&
-                        (config.tablesWhitelist == null || config.tablesWhitelist.contains(tableName))) {
+                if (!acceptTable(tableName)) {
                     ignoredTables.add(tableName);
                     continue;
                 }
@@ -126,6 +124,12 @@ public void start(Map<String, String> props) {
         streamsClient.shutdown();
     }
 
+    private boolean acceptTable(String tableName) {
+        return tableName.startsWith(config.tablesPrefix)
+                && (config.tablesWhitelist.isEmpty() || config.tablesWhitelist.contains(tableName))
+                && !config.tablesBlacklist.contains(tableName);
+    }
+
     @Override
     public void stop() {
     }