Merge pull request #103 from data-integrations/bugfix_release/CDAP-17370

CDAP-17370 Provide unique consumer id to each worker instance in the pipeline for mysql source.

Author: sagarkapare

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

@@ -40,7 +40,9 @@ 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.")
+  @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;
 
   @Description("Database to consume events for.")

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

@@ -110,7 +110,7 @@ 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())
+      .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