add debezium config to avoid unnecessary ddl event (#167)

Author: seanzhougoogle

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

@@ -117,12 +117,13 @@ public void start(Offset offset) {
       .with("database.user", config.getUser())
       .with("database.password", config.getPassword())
       .with("database.history", DBSchemaHistory.class.getName())
-      .with("database.whitelist", config.getDatabase())
+      .with("database.include.list", 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()))
+      .with("database.history.store.only.monitored.tables.ddl", true)
+      .with("table.include.list", String.join(",", sourceTableMap.keySet()))
       .with("snapshot.mode", config.getReplicateExistingData() ? "initial" : "schema_only")
-      .with(MySqlConstantOffsetBackingStore.REPLICATION_CONNECTOR_NAME, replicationConnectorName);;
+      .with(MySqlConstantOffsetBackingStore.REPLICATION_CONNECTOR_NAME, replicationConnectorName);
 
     if (config.getConsumerID() != null) {
       // If not provided debezium will randomly pick integer between 5400 and 6400.
@@ -135,6 +136,7 @@ public void start(Offset offset) {
     }
 
     Configuration debeziumConf = configBuilder.build();
+    LOG.info("Debezium configuration : {}", debeziumConf);
     MySqlConnectorConfig mysqlConf = new MySqlConnectorConfig(debeziumConf);
     DBSchemaHistory.deltaRuntimeContext = context;
     /*

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

@@ -113,7 +113,9 @@ public void accept(SourceRecord sourceRecord) {
     } catch (IOException e) {
       LOG.warn("Unable to set source state to OK.", e);
     }
-
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("Receiving source record {}", sourceRecord);
+    }
     if (sourceRecord.value() == null) {
       return;
     }