Merge branch 'prb-its' into my-test

Author: sumitjnn

.github/workflows/build.yml

@@ -47,8 +47,8 @@ jobs:
         run: which docker && docker ps
 
       - name: Build with Maven
-        run: mvn clean verify -fae -T 2 -B -V -DcloudBuild -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon
-             .httpconnectionManager.ttlSeconds=25
+        run: mvn clean verify -fae -T 2 -B -V -DcloudBuild -Dmaven.wagon.http.retryHandler.count=3
+          -Dmaven.wagon.httpconnectionManager.ttlSeconds=25
 
       - name: Archive build artifacts
         uses: actions/upload-artifact@v3

mysql-delta-plugins/pom.xml

@@ -53,7 +53,7 @@
     <dependency>
       <groupId>mysql</groupId>
       <artifactId>mysql-connector-java</artifactId>
-      <version>8.0.19</version>
+      <version>8.0.28</version>
       <scope>test</scope>
     </dependency>
   </dependencies>

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

@@ -30,6 +30,7 @@
 import io.cdap.delta.plugin.mock.BlockingEventEmitter;
 import io.cdap.delta.plugin.mock.MockContext;
 import io.cdap.delta.plugin.mock.MockEventEmitter;
+import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
@@ -41,6 +42,7 @@
 import java.sql.Date;
 import java.sql.DriverManager;
 import java.sql.PreparedStatement;
+import java.sql.SQLException;
 import java.sql.Statement;
 import java.time.LocalDate;
 import java.util.Collections;
@@ -57,7 +59,7 @@
  * are some classloading issues due to copied debezium classes.
  */
 public class MySqlEventReaderIntegrationTest {
-  private static final String DB = "test";
+  private static final int CONSUMER_ID = 13;
   private static final String CUSTOMERS_TABLE = "customers";
   private static final Schema CUSTOMERS_SCHEMA = Schema.recordOf(
     "customers",
@@ -69,9 +71,14 @@ public class MySqlEventReaderIntegrationTest {
     BINARYCOL_TABLE,
     Schema.Field.of("id", Schema.of(Schema.Type.INT)),
     Schema.Field.of("bincol", Schema.of(Schema.Type.STRING)));
+  private static final String HOST = "localhost";
+  private static final String DB = "test";
+  private static final String USER = "root";
 
   private static String password;
   private static int port;
+  private static Properties connProperties;
+  private static String connectionUrl;
 
 
   @BeforeClass
@@ -84,10 +91,10 @@ public static void setupClass() throws Exception {
     }
     port = Integer.parseInt(properties.getProperty("mysql.port"));
 
-    Properties connProperties = new Properties();
-    connProperties.put("user", "root");
+    connProperties = new Properties();
+    connProperties.put("user", USER);
     connProperties.put("password", password);
-    String connectionUrl = String.format("jdbc:mysql://localhost:%d", port);
+    connectionUrl = String.format("jdbc:mysql://localhost:%d", port);
     DriverManager.getDriver(connectionUrl);
 
     // wait until a connection can be established
@@ -146,14 +153,24 @@ public static void setupClass() throws Exception {
     }
   }
 
+  @AfterClass
+  public static void tearDown() throws SQLException {
+    // drop database
+    try (Connection connection = DriverManager.getConnection(connectionUrl, connProperties)) {
+      try (Statement statement = connection.createStatement()) {
+        statement.execute("DROP DATABASE " + DB);
+      }
+    }
+  }
+
   @Test
   public void test() throws InterruptedException {
     SourceTable sourceTable = new SourceTable(DB, CUSTOMERS_TABLE, null,
                                               Collections.emptySet(), Collections.emptySet(), Collections.emptySet());
 
     DeltaSourceContext context = new MockContext(Driver.class);
-    MockEventEmitter eventEmitter = new MockEventEmitter(6);
-    MySqlConfig config = new MySqlConfig("localhost", port, "root", password, 13, DB,
+    MockEventEmitter eventEmitter = new MockEventEmitter(7);
+    MySqlConfig config = new MySqlConfig(HOST, port, USER, password, CONSUMER_ID, DB,
                                          TimeZone.getDefault().getID());
 
     MySqlEventReader eventReader = new MySqlEventReader(Collections.singleton(sourceTable), config,
@@ -164,7 +181,7 @@ public void test() throws InterruptedException {
     eventEmitter.waitForExpectedEvents(30, TimeUnit.SECONDS);
 
     Assert.assertEquals(4, eventEmitter.getDdlEvents().size());
-    Assert.assertEquals(2, eventEmitter.getDmlEvents().size());
+    Assert.assertEquals(3, eventEmitter.getDmlEvents().size());
 
     DDLEvent ddlEvent = eventEmitter.getDdlEvents().get(0);
     Assert.assertEquals(DDLOperation.Type.DROP_TABLE, ddlEvent.getOperation().getType());
@@ -184,14 +201,20 @@ public void test() throws InterruptedException {
     Assert.assertEquals(DB, ddlEvent.getOperation().getDatabaseName());
     Assert.assertEquals(CUSTOMERS_TABLE, ddlEvent.getOperation().getTableName());
     Assert.assertEquals(Collections.singletonList("id"), ddlEvent.getPrimaryKey());
+
     Assert.assertEquals(CUSTOMERS_SCHEMA, ddlEvent.getSchema());
 
     DMLEvent dmlEvent = eventEmitter.getDmlEvents().get(0);
     Assert.assertEquals(DMLOperation.Type.INSERT, dmlEvent.getOperation().getType());
     Assert.assertEquals(DB, dmlEvent.getOperation().getDatabaseName());
     Assert.assertEquals(CUSTOMERS_TABLE, dmlEvent.getOperation().getTableName());
     StructuredRecord row = dmlEvent.getRow();
-    StructuredRecord expected = StructuredRecord.builder(CUSTOMERS_SCHEMA)
+
+    // Take schema name from the row as it is generated by debezium
+    // so it can be different from the one in our schema but does not impact data
+    Schema expectedSchema = Schema.recordOf(row.getSchema().getRecordName(), CUSTOMERS_SCHEMA.getFields());
+
+    StructuredRecord expected = StructuredRecord.builder(expectedSchema)
       .set("id", 0)
       .set("name", "alice")
       .setDate("bday", LocalDate.ofEpochDay(0))
@@ -203,7 +226,7 @@ public void test() throws InterruptedException {
     Assert.assertEquals(DB, dmlEvent.getOperation().getDatabaseName());
     Assert.assertEquals(CUSTOMERS_TABLE, dmlEvent.getOperation().getTableName());
     row = dmlEvent.getRow();
-    expected = StructuredRecord.builder(CUSTOMERS_SCHEMA)
+    expected = StructuredRecord.builder(expectedSchema)
       .set("id", 1)
       .set("name", "bob")
       .setDate("bday", LocalDate.ofEpochDay(365))
@@ -215,7 +238,7 @@ public void test() throws InterruptedException {
     Assert.assertEquals(DB, dmlEvent.getOperation().getDatabaseName());
     Assert.assertEquals(CUSTOMERS_TABLE, dmlEvent.getOperation().getTableName());
     row = dmlEvent.getRow();
-    expected = StructuredRecord.builder(CUSTOMERS_SCHEMA)
+    expected = StructuredRecord.builder(expectedSchema)
       .set("id", 2)
       .set("name", "tim")
       .setDate("bday", null)
@@ -232,7 +255,7 @@ public void stopReaderTest() throws Exception {
     BlockingQueue<DDLEvent> ddlEvents = new ArrayBlockingQueue<>(1);
     BlockingQueue<DMLEvent> dmlEvents = new ArrayBlockingQueue<>(1);
     EventEmitter eventEmitter = new BlockingEventEmitter(ddlEvents, dmlEvents);
-    MySqlConfig config = new MySqlConfig("localhost", port, "root", password, 13, DB,
+    MySqlConfig config = new MySqlConfig(HOST, port, USER, password, CONSUMER_ID, DB,
                                          TimeZone.getDefault().getID());
 
     MySqlEventReader eventReader = new MySqlEventReader(Collections.singleton(sourceTable), config,
@@ -264,7 +287,7 @@ public void testBinaryHandlingModebyDebezium() throws InterruptedException {
       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,
+      MySqlConfig config = new MySqlConfig(HOST, port, USER, password, CONSUMER_ID, DB,
                                            TimeZone.getDefault().getID());
 
       MySqlEventReader eventReader = new MySqlEventReader(Collections.singleton(sourceTable), config,

mysql-delta-plugins/src/test/resources/logback-test.xml

@@ -0,0 +1,35 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<!--
+  ~ Copyright © 2022 Cask Data, Inc.
+  ~
+  ~ Licensed under the Apache License, Version 2.0 (the "License"); you may not
+  ~ use this file except in compliance with the License. You may obtain a copy of
+  ~ the License at
+  ~
+  ~ http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+  ~ WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+  ~ License for the specific language governing permissions and limitations under
+  ~ the License.
+  -->
+
+<configuration>
+
+  <appender name="Console" class="ch.qos.logback.core.ConsoleAppender">
+    <encoder>
+      <pattern>%d{ISO8601} - %-5p [%t:%C{1}@%L] - %m%n</pattern>
+    </encoder>
+  </appender>
+
+  <logger name="io.debezium.connector" level="WARN">
+    <appender-ref ref="Console"/>
+  </logger>
+
+  <root level="INFO">
+    <appender-ref ref="Console"/>
+  </root>
+
+</configuration>