CDAP-17582 Allow passing the debezium connector configurations through runtime arguments.

Author: sagarkapare

delta-plugins-common/src/main/java/io/cdap/delta/plugin/common/RuntimeArguments.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright © 2020 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.delta.plugin.common;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Utility class for manipulating runtime arguments.
+ */
+public final class RuntimeArguments {
+  private RuntimeArguments() {
+  }
+
+  /**
+   * Identifies arguments with a given prefix, removes them from the arguments and adds new arguments back without
+   * the prefix. Arguments which do not start with the given prefix will be ignored.
+   *
+   * @param prefix prefix to look for in arguments
+   * @param arguments the runtime arguments
+   * @return a new map that contains the arguments without prefix
+   */
+  public static Map<String, String> extractPrefixed(String prefix, Map<String, String> arguments) {
+    Map<String, String> result = new HashMap<>();
+    for (Map.Entry<String, String> entry : arguments.entrySet()) {
+      if (entry.getKey().startsWith(prefix)) {
+        result.put(entry.getKey().substring(prefix.length()), entry.getValue());
+      }
+    }
+    return result;
+  }
+}

delta-plugins-common/src/test/java/io/cdap/delta/plugin/common/RuntimeArgumentsTest.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright © 2020 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.delta.plugin.common;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Map;
+
+/**
+ * Test for RuntimeArguments.
+ */
+public class RuntimeArgumentsTest {
+
+  @Test
+  public void testRemovePrefix() {
+    Map<String, String> arguments = ImmutableMap.of("source.connector.database.host", "1.1.1.1",
+                                                    "source.connector.name", "somename",
+                                                    "somekey", "somevalue");
+    Map<String, String> expected = ImmutableMap.of("database.host", "1.1.1.1",
+                                                   "name", "somename");
+
+    Assert.assertEquals(expected, RuntimeArguments.extractPrefixed("source.connector.", arguments));
+  }
+}

mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlEventReader.java

@@ -24,6 +24,7 @@
 import io.cdap.delta.api.SourceTable;
 import io.cdap.delta.plugin.common.DBSchemaHistory;
 import io.cdap.delta.plugin.common.NotifyingCompletionCallback;
+import io.cdap.delta.plugin.common.RuntimeArguments;
 import io.debezium.DebeziumException;
 import io.debezium.config.CommonConnectorConfig;
 import io.debezium.config.Configuration;
@@ -58,11 +59,13 @@
  */
 public class MySqlEventReader implements EventReader {
   public static final Logger LOG = LoggerFactory.getLogger(MySqlEventReader.class);
+  private static final String SOURCE_CONNECTOR_PREFIX = "source.connector.";
   private final MySqlConfig config;
   private final EventEmitter emitter;
   private final ExecutorService executorService;
   private final DeltaSourceContext context;
   private final Set<SourceTable> sourceTables;
+  private final Map<String, String> debeziumConnectorConfigs;
   private EmbeddedEngine engine;
   private volatile boolean failedToStop;
 
@@ -74,6 +77,8 @@ public MySqlEventReader(Set<SourceTable> sourceTables, MySqlConfig config,
     this.emitter = emitter;
     this.executorService = Executors.newSingleThreadScheduledExecutor();
     this.failedToStop = false;
+    this.debeziumConnectorConfigs = RuntimeArguments.extractPrefixed(SOURCE_CONNECTOR_PREFIX,
+                                                                     context.getRuntimeArguments());
   }
 
   @Override
@@ -121,6 +126,11 @@ public void start(Offset offset) {
       configBuilder = configBuilder.with("database.server.id", config.getConsumerID() + context.getInstanceId());
     }
 
+    LOG.info("Overriding mysql connector configs with arguments {}", debeziumConnectorConfigs);
+    for (Map.Entry<String, String> entry: debeziumConnectorConfigs.entrySet()) {
+      configBuilder = configBuilder.with(entry.getKey(), entry.getValue());
+    }
+
     Configuration debeziumConf = configBuilder.build();
     MySqlConnectorConfig mysqlConf = new MySqlConnectorConfig(debeziumConf);
     DBSchemaHistory.deltaRuntimeContext = context;

sqlserver-delta-plugins/src/main/java/io/cdap/delta/sqlserver/SqlServerEventReader.java

@@ -24,6 +24,7 @@
 import io.cdap.delta.api.SourceTable;
 import io.cdap.delta.plugin.common.DBSchemaHistory;
 import io.cdap.delta.plugin.common.NotifyingCompletionCallback;
+import io.cdap.delta.plugin.common.RuntimeArguments;
 import io.debezium.config.Configuration;
 import io.debezium.connector.sqlserver.SourceInfo;
 import io.debezium.connector.sqlserver.SqlServerConnection;
@@ -52,11 +53,13 @@
  */
 public class SqlServerEventReader implements EventReader {
   private static final Logger LOG = LoggerFactory.getLogger(SqlServerEventReader.class);
+  private static final String SOURCE_CONNECTOR_PREFIX = "source.connector.";
   private final SqlServerConfig config;
   private final EventEmitter emitter;
   private final DeltaSourceContext context;
   private final ExecutorService executorService;
   private final Set<SourceTable> tables;
+  private final Map<String, String> debeziumConnectorConfigs;
   private volatile boolean failedStopping;
   private EmbeddedEngine engine;
 
@@ -68,6 +71,8 @@ public SqlServerEventReader(Set<SourceTable> tables, SqlServerConfig config,
     this.tables = tables;
     this.executorService = Executors.newSingleThreadScheduledExecutor();
     this.failedStopping = false;
+    this.debeziumConnectorConfigs = RuntimeArguments.extractPrefixed(SOURCE_CONNECTOR_PREFIX,
+                                                                     context.getRuntimeArguments());
   }
 
   @Override
@@ -101,7 +106,7 @@ public void start(Offset offset) {
     Map<String, String> state = offset.get(); // this will never be null
     // offset config
     String isSnapshotCompleted = state.getOrDefault(SqlServerConstantOffsetBackingStore.SNAPSHOT_COMPLETED, "");
-    Configuration debeziumConf = Configuration.create()
+    Configuration.Builder configBuilder = Configuration.create()
       .with("connector.class", SqlServerConnector.class.getName())
       .with("offset.storage", SqlServerConstantOffsetBackingStore.class.getName())
       .with("offset.flush.interval.ms", 1000)
@@ -120,8 +125,14 @@ public void start(Offset offset) {
       .with("database.dbname", databaseName)
       .with("table.whitelist", String.join(",", sourceTableMap.keySet()))
       .with("database.server.name", "dummy") // this is the kafka topic for hosted debezium - it doesn't matter
-      .with("database.serverTimezone", config.getServerTimezone())
-      .build();
+      .with("database.serverTimezone", config.getServerTimezone());
+
+    LOG.info("Overriding sql server connector configs with arguments {}", debeziumConnectorConfigs);
+    for (Map.Entry<String, String> entry: debeziumConnectorConfigs.entrySet()) {
+      configBuilder = configBuilder.with(entry.getKey(), entry.getValue());
+    }
+
+    Configuration debeziumConf = configBuilder.build();
 
     DBSchemaHistory.deltaRuntimeContext = context;