Merge pull request #122 from data-integrations/fix/make-consumer-id-optional

CDAP-17472 Make consumer id optional for mysql plugin.

Author: sagarkapare

mysql-delta-plugins/docs/mysql-cdcSource.md

@@ -58,9 +58,9 @@ Plugin Properties
 
 **Port:** Port to use to connect to the MySQL server.
 
-**Consumer ID:** An unique numeric ID to identify this origin as an event consumer. This number cannot be the same as 
-another Delta Replicator that is reading from the server, and it cannot be the same as the server-id for any MySQL 
-slave that is replicating from the server.
+**Consumer ID:** Optional unique numeric ID to identify this origin as an event consumer. This number cannot be the 
+same as another Delta Replicator that is reading from the server, and it cannot be the same as the server-id for any
+ MySQL slave that is replicating from the server. By default, random number will be used. 
 
 **Server Timezone:** Timezone of the MySQL server. This is used when converting dates into timestamps.
 

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

@@ -41,10 +41,11 @@ public class MySqlConfig extends PluginConfig {
   @Description("Password to use to connect to the MySQL server.")
   private String password;
 
-  @Description("A unique numeric ID to identify this origin as an event consumer. When replication pipeline is " +
-    "configured with multiple instances, each instance gets unique consumer id by adding instance id to " +
-    "this supplied consumer id.")
-  private int consumerID;
+  @Nullable
+  @Description("An optional unique numeric ID to identify this origin as an event consumer. When replication " +
+    "pipeline is configured with multiple instances, each instance gets unique consumer id by adding instance id to " +
+    "this supplied consumer id. By default, random number will be used.")
+  private Integer consumerID;
 
   @Description("Database to replicate data from.")
   private String database;
@@ -83,7 +84,8 @@ public String getPassword() {
     return password;
   }
 
-  public int getConsumerID() {
+  @Nullable
+  public Integer getConsumerID() {
     return consumerID;
   }
 

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

@@ -93,7 +93,7 @@ public void start(Offset offset) {
     Map<String, SourceTable> sourceTableMap = sourceTables.stream().collect(
       Collectors.toMap(t -> config.getDatabase() + "." + t.getTable(), t -> t));
     Map<String, String> state = offset.get(); // state map is always not null
-    Configuration debeziumConf = Configuration.create()
+    Configuration.Builder configBuilder = Configuration.create()
       .with("connector.class", MySqlConnector.class.getName())
       .with("offset.storage", MySqlConstantOffsetBackingStore.class.getName())
       .with("offset.flush.interval.ms", 1000)
@@ -110,13 +110,18 @@ public void start(Offset offset) {
       .with("database.port", config.getPort())
       .with("database.user", config.getUser())
       .with("database.password", config.getPassword())
-      .with("database.server.id", config.getConsumerID() + context.getInstanceId())
       .with("database.history", DBSchemaHistory.class.getName())
       .with("database.whitelist", config.getDatabase())
       .with("database.server.name", "dummy") // this is the kafka topic for hosted debezium - it doesn't matter
       .with("database.serverTimezone", config.getServerTimezone())
-      .with("table.whitelist", String.join(",", sourceTableMap.keySet()))
-      .build();
+      .with("table.whitelist", String.join(",", sourceTableMap.keySet()));
+
+    if (config.getConsumerID() != null) {
+      // If not provided debezium will randomly pick integer between 5400 and 6400.
+      configBuilder = configBuilder.with("database.server.id", config.getConsumerID() + context.getInstanceId());
+    }
+
+    Configuration debeziumConf = configBuilder.build();
     MySqlConnectorConfig mysqlConf = new MySqlConnectorConfig(debeziumConf);
     DBSchemaHistory.deltaRuntimeContext = context;
     /*

mysql-delta-plugins/widgets/mysql-cdcSource.json

@@ -20,11 +20,6 @@
             "default": "3306"
           }
         },
-        {
-          "name": "consumerID",
-          "label": "Consumer ID",
-          "widget-type": "textbox"
-        },
         {
           "name": "jdbcPluginName",
           "label": "JDBC Plugin Name",
@@ -58,6 +53,11 @@
     {
       "label": "Advanced",
       "properties": [
+        {
+          "name": "consumerID",
+          "label": "Consumer ID",
+          "widget-type": "textbox"
+        },
         {
           "name": "serverTimezone",
           "label": "Server Timezone",