Merge pull request #261 from gkstechie/master

topics.regex support in BigQuery Sink Connector

Author: whynick1

kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/config/BigQuerySinkConfig.java

@@ -66,8 +66,21 @@ public class BigQuerySinkConfig extends AbstractConfig {
   private static final ConfigDef.Width TOPICS_WIDTH =            ConfigDef.Width.LONG;
   private static final String TOPICS_DOC =
       "List of topics to consume, separated by commas";
+  public static final String TOPICS_DEFAULT = "";
   private static final String TOPICS_DISPLAY =                   "Topics";
 
+  public static final String TOPICS_REGEX_CONFIG =                     "topics.regex";
+  private static final ConfigDef.Type TOPICS_REGEX_TYPE =              ConfigDef.Type.STRING;
+  private static final ConfigDef.Importance TOPICS_REGEX_IMPORTANCE =  ConfigDef.Importance.HIGH;
+  private static final String TOPICS_REGEX_GROUP =                     "Common";
+  private static final int TOPICS_REGEX_ORDER_IN_GROUP =               4;
+  private static final ConfigDef.Width TOPICS_REGEX_WIDTH =            ConfigDef.Width.LONG;
+  private static final String TOPICS_REGEX_DOC = "Regular expression giving topics to consume. " +
+	  "Under the hood, the regex is compiled to a <code>java.util.regex.Pattern</code>. " +
+	  "Only one of " + TOPICS_CONFIG + " or " + TOPICS_REGEX_CONFIG + " should be specified.";
+  public static final String TOPICS_REGEX_DEFAULT = "";
+  private static final String TOPICS_REGEX_DISPLAY = "Topics regex";
+
   public static final String ENABLE_BATCH_CONFIG =                         "enableBatchLoad";
   private static final ConfigDef.Type ENABLE_BATCH_TYPE =                  ConfigDef.Type.LIST;
   private static final List<String> ENABLE_BATCH_DEFAULT =                 Collections.emptyList();
@@ -225,12 +238,23 @@ public class BigQuerySinkConfig extends AbstractConfig {
         .define(
             TOPICS_CONFIG,
             TOPICS_TYPE,
+            TOPICS_DEFAULT,
             TOPICS_IMPORTANCE,
             TOPICS_DOC,
             TOPICS_GROUP,
             TOPICS_ORDER_IN_GROUP,
             TOPICS_WIDTH,
             TOPICS_DISPLAY)
+        .define(
+            TOPICS_REGEX_CONFIG,
+            TOPICS_REGEX_TYPE,
+            TOPICS_REGEX_DEFAULT,
+            TOPICS_REGEX_IMPORTANCE,
+            TOPICS_REGEX_DOC,
+            TOPICS_REGEX_GROUP,
+            TOPICS_REGEX_ORDER_IN_GROUP,
+            TOPICS_REGEX_WIDTH,
+            TOPICS_REGEX_DISPLAY)
         .define(
             ENABLE_BATCH_CONFIG,
             ENABLE_BATCH_TYPE,
@@ -343,6 +367,34 @@ public class BigQuerySinkConfig extends AbstractConfig {
             TABLE_CREATE_DOC
         );
   }
+    /**
+     * Throw an exception if the passed-in properties do not constitute a valid sink.
+     * @param props sink configuration properties
+     */
+    public static void validate(Map<String, String> props) {
+        final boolean hasTopicsConfig = hasTopicsConfig(props);
+        final boolean hasTopicsRegexConfig = hasTopicsRegexConfig(props);
+
+        if (hasTopicsConfig && hasTopicsRegexConfig) {
+            throw new ConfigException(TOPICS_CONFIG + " and " + TOPICS_REGEX_CONFIG +
+                " are mutually exclusive options, but both are set.");
+        }
+
+        if (!hasTopicsConfig && !hasTopicsRegexConfig) {
+            throw new ConfigException("Must configure one of " +
+                TOPICS_CONFIG + " or " + TOPICS_REGEX_CONFIG);
+        }
+    }
+
+    public static boolean hasTopicsConfig(Map<String, String> props) {
+        String topicsStr = props.get(TOPICS_CONFIG);
+        return topicsStr != null && !topicsStr.trim().isEmpty();
+    }
+
+    public static boolean hasTopicsRegexConfig(Map<String, String> props) {
+        String topicsRegexStr = props.get(TOPICS_REGEX_CONFIG);
+        return topicsRegexStr != null && !topicsRegexStr.trim().isEmpty();
+    }
 
   @SuppressWarnings("unchecked")
   public static class Validator implements ConfigDef.Validator {

kcbq-connector/src/test/java/com/wepay/kafka/connect/bigquery/BigQuerySinkConnectorTest.java

@@ -40,6 +40,7 @@
 
 import com.wepay.kafka.connect.bigquery.exception.BigQueryConnectException;
 import com.wepay.kafka.connect.bigquery.exception.SinkConfigConnectException;
+import org.apache.kafka.common.config.ConfigException;
 
 import org.apache.kafka.connect.data.Schema;
 
@@ -131,9 +132,14 @@ public void testConfig() {
   // Make sure that a config exception is properly translated into a SinkConfigConnectException
   @Test(expected = SinkConfigConnectException.class)
   public void testConfigException() {
-    Map<String, String> badProperties = propertiesFactory.getProperties();
-    badProperties.remove(BigQuerySinkConfig.TOPICS_CONFIG);
-    new BigQuerySinkConnector().start(badProperties);
+    try {
+      Map<String, String> badProperties = propertiesFactory.getProperties();
+      badProperties.remove(BigQuerySinkConfig.TOPICS_CONFIG);
+      BigQuerySinkConfig.validate(badProperties);
+      new BigQuerySinkConnector().start(badProperties);
+    } catch (ConfigException e) {
+      throw new SinkConfigConnectException(e);
+    }
   }
 
   @Test

kcbq-connector/src/test/java/com/wepay/kafka/connect/bigquery/BigQuerySinkTaskTest.java

@@ -41,6 +41,7 @@
 import com.wepay.kafka.connect.bigquery.config.BigQuerySinkTaskConfig;
 import com.wepay.kafka.connect.bigquery.exception.BigQueryConnectException;
 import com.wepay.kafka.connect.bigquery.exception.SinkConfigConnectException;
+import org.apache.kafka.common.config.ConfigException;
 
 import org.apache.kafka.common.record.TimestampType;
 import org.apache.kafka.connect.data.Schema;
@@ -490,15 +491,20 @@ public void testInterruptedException() {
   // Make sure that a ConfigException is properly translated into a SinkConfigConnectException
   @Test(expected = SinkConfigConnectException.class)
   public void testConfigException() {
-    Map<String, String> badProperties = propertiesFactory.getProperties();
-    badProperties.remove(BigQuerySinkConfig.TOPICS_CONFIG);
-
-    SchemaRetriever schemaRetriever = mock(SchemaRetriever.class);
-    SchemaManager schemaManager = mock(SchemaManager.class);
-
-    BigQuerySinkTask testTask =
-        new BigQuerySinkTask(mock(BigQuery.class), schemaRetriever, mock(Storage.class), schemaManager);
-    testTask.start(badProperties);
+    try {
+      Map<String, String> badProperties = propertiesFactory.getProperties();
+      badProperties.remove(BigQuerySinkConfig.TOPICS_CONFIG);
+      BigQuerySinkConfig.validate(badProperties);
+  
+      SchemaRetriever schemaRetriever = mock(SchemaRetriever.class);
+      SchemaManager schemaManager = mock(SchemaManager.class);
+  
+      BigQuerySinkTask testTask =
+          new BigQuerySinkTask(mock(BigQuery.class), schemaRetriever, mock(Storage.class), schemaManager);
+      testTask.start(badProperties);
+    } catch (ConfigException e) {
+      throw new SinkConfigConnectException(e);
+    }
   }
 
   @Test