Merge pull request #214 from data-integrations/CDAP-20121

MYSQL Create Table doesn't pick up the schema according to debezium's…

Author: sahusanket

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

@@ -36,9 +36,11 @@
 public class MockContext implements DeltaSourceContext {
   private final Class<?> driverClass;
   private final Map<String, byte[]> state = new HashMap<>();
+  private Map<String, String> runtimeArguments;
 
   public MockContext(Class<?> driverClass) {
     this.driverClass = driverClass;
+    this.runtimeArguments = new HashMap<>();
   }
 
   @Override
@@ -88,7 +90,7 @@ public Map<String, String> getTags() {
 
   @Override
   public Map<String, String> getRuntimeArguments() {
-    return new HashMap<>();
+    return this.runtimeArguments;
   }
 
   @Override
@@ -151,4 +153,8 @@ public <T> T newPluginInstance(String pluginId, MacroEvaluator evaluator) {
   public void notifyFailed(Throwable throwable) {
     // no-op
   }
+
+  public void addRuntimeArgument(String key, String value) {
+    this.runtimeArguments.put(key, value);
+  }
 }

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

@@ -59,7 +59,7 @@
  */
 public class MySqlEventReader implements EventReader {
   public static final Logger LOG = LoggerFactory.getLogger(MySqlEventReader.class);
-  private static final String SOURCE_CONNECTOR_PREFIX = "source.connector.";
+  protected static final String SOURCE_CONNECTOR_PREFIX = "source.connector.";
   private final MySqlConfig config;
   private final EventEmitter emitter;
   private final ExecutorService executorService;
@@ -200,7 +200,7 @@ private static MySqlValueConverters getValueConverters(MySqlConnectorConfig conf
 
     boolean timeAdjusterEnabled = configuration.getConfig().getBoolean(MySqlConnectorConfig.ENABLE_TIME_ADJUSTER);
     return new MySqlValueConverters(decimalMode, timePrecisionMode, bigIntUnsignedMode,
-                                    CommonConnectorConfig.BinaryHandlingMode.BYTES,
+                                    configuration.binaryHandlingMode(),
                                     timeAdjusterEnabled ? MySqlEventReader::adjustTemporal : x -> x,
                                     (message, exception) -> {
       throw new DebeziumException(message, exception);

mysql-delta-plugins/src/test/java/io/cdap/delta/mysql/MySqlEventReaderIntegrationTest.java

@@ -64,6 +64,12 @@ public class MySqlEventReaderIntegrationTest {
     Schema.Field.of("id", Schema.of(Schema.Type.INT)),
     Schema.Field.of("name", Schema.of(Schema.Type.STRING)),
     Schema.Field.of("bday", Schema.nullableOf(Schema.of(Schema.LogicalType.DATE))));
+  private static final String BINARYCOL_TABLE = "binarycoltable";
+  private static final Schema EXPECTED_BINARYCOL_SCHEMA = Schema.recordOf(
+    BINARYCOL_TABLE,
+    Schema.Field.of("id", Schema.of(Schema.Type.INT)),
+    Schema.Field.of("bincol", Schema.of(Schema.Type.STRING)));
+
   private static String password;
   private static int port;
 
@@ -109,6 +115,13 @@ public static void setupClass() throws Exception {
                         CUSTOMERS_TABLE));
       }
 
+      // create table with Binary col
+      try (Statement statement = connection.createStatement()) {
+        statement.execute(
+          String.format("CREATE TABLE %s (id int PRIMARY KEY, bincol BINARY(16) not null)",
+                        BINARYCOL_TABLE));
+      }
+
       // insert sample data
       try (PreparedStatement ps = connection.prepareStatement(String.format("INSERT INTO %s VALUES (?, ?, ?)",
                                                                             CUSTOMERS_TABLE))) {
@@ -239,4 +252,30 @@ public void stopReaderTest() throws Exception {
     eventReader.stop();
     Assert.assertFalse(eventReader.failedToStop());
   }
-}
+
+    @Test
+    public void testBinaryHandlingModebyDebezium() throws InterruptedException {
+      //https://debezium.io/documentation/reference/stable/connectors/mysql.html#mysql-property-binary-handling-mode
+      SourceTable sourceTable = new SourceTable(DB, BINARYCOL_TABLE, null,
+                                                Collections.emptySet(), Collections.emptySet(), Collections.emptySet());
+
+      MockContext context = new MockContext(Driver.class);
+      context.addRuntimeArgument(MySqlEventReader.SOURCE_CONNECTOR_PREFIX +  "binary.handling.mode" , "HEX");
+
+      MockEventEmitter eventEmitter = new MockEventEmitter(4);
+      MySqlConfig config = new MySqlConfig("localhost", port, "root", password, 13, DB,
+                                           TimeZone.getDefault().getID());
+
+      MySqlEventReader eventReader = new MySqlEventReader(Collections.singleton(sourceTable), config,
+                                                          context, eventEmitter);
+      eventReader.start(new Offset());
+
+      eventEmitter.waitForExpectedEvents(30, TimeUnit.SECONDS);
+
+      DDLEvent ddlEvent = eventEmitter.getDdlEvents().get(3);
+      Assert.assertEquals(DDLOperation.Type.CREATE_TABLE, ddlEvent.getOperation().getType());
+      Assert.assertEquals(DB, ddlEvent.getOperation().getDatabaseName());
+      Assert.assertEquals(BINARYCOL_TABLE, ddlEvent.getOperation().getTableName());
+      Assert.assertEquals(EXPECTED_BINARYCOL_SCHEMA, ddlEvent.getSchema());
+    }
+  }