Merge pull request #109 from data-integrations/bugfix_release/upgrade-debezium-version

[CDAP-17402] Upgrade debizum version to 1.3.1.Final

Author: sagarkapare

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

@@ -78,6 +78,11 @@ public synchronized boolean exists() {
     }
   }
 
+  @Override
+  public boolean storageExists() {
+    return true;
+  }
+
   // TODO: cache history, should only have to read once
   private List<HistoryRecord> getHistory() {
     List<HistoryRecord> history = new ArrayList<>();

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

@@ -304,7 +304,8 @@ public static Schema convert(org.apache.kafka.connect.data.Schema schema) {
         for (Field field : fields) {
           cdapFields.add(Schema.Field.of(field.name(), convert(field.schema())));
         }
-        converted = Schema.recordOf(schema.name(), cdapFields);
+
+        converted = Schema.recordOf(schema.name() == null ? "dummy.schema.name" : schema.name(), cdapFields);
         break;
       default:
         // should never happen, all values are listed above

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

@@ -25,11 +25,14 @@
 import io.cdap.delta.api.SourceTable;
 import io.cdap.delta.plugin.common.DBSchemaHistory;
 import io.cdap.delta.plugin.common.NotifyingCompletionCallback;
+import io.debezium.DebeziumException;
+import io.debezium.config.CommonConnectorConfig;
 import io.debezium.config.Configuration;
 import io.debezium.connector.mysql.MySqlConnector;
 import io.debezium.connector.mysql.MySqlConnectorConfig;
 import io.debezium.connector.mysql.MySqlJdbcContext;
 import io.debezium.connector.mysql.MySqlValueConverters;
+import io.debezium.connector.mysql.antlr.MySqlAntlrDdlParser;
 import io.debezium.embedded.EmbeddedEngine;
 import io.debezium.jdbc.JdbcConnection;
 import io.debezium.jdbc.JdbcValueConverters;
@@ -132,7 +135,7 @@ public void start(Offset offset) {
     }
 
     MySqlValueConverters mySqlValueConverters = getValueConverters(mysqlConf);
-    DdlParser ddlParser = mysqlConf.getDdlParsingMode().getNewParserInstance(mySqlValueConverters, tableId -> true);
+    DdlParser ddlParser = new MySqlAntlrDdlParser(mySqlValueConverters, tableId -> true);
 
     ClassLoader oldCL = Thread.currentThread().getContextClassLoader();
     Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
@@ -179,7 +182,11 @@ private static MySqlValueConverters getValueConverters(MySqlConnectorConfig conf
 
     boolean timeAdjusterEnabled = configuration.getConfig().getBoolean(MySqlConnectorConfig.ENABLE_TIME_ADJUSTER);
     return new MySqlValueConverters(decimalMode, timePrecisionMode, bigIntUnsignedMode,
-                                    timeAdjusterEnabled ? MySqlEventReader::adjustTemporal : x -> x);
+                                    CommonConnectorConfig.BinaryHandlingMode.BYTES,
+                                    timeAdjusterEnabled ? MySqlEventReader::adjustTemporal : x -> x,
+                                    (message, exception) -> {
+      throw new DebeziumException(message, exception);
+    });
   }
 
   private static Temporal adjustTemporal(Temporal temporal) {