[Don't Merge] More Test Fixes

src/test/java/io/debezium/connector/db2/Db2ConnectorIT.java

@@ -58,9 +58,15 @@ public class Db2ConnectorIT extends AbstractConnectorTest {
 
     @Before
     public void before() throws SQLException {
-        TestHelper.dropAllTables();
+        // CHECKSTYLE:OFF
+        System.out.println("failFlakyTests: " + System.getProperty(Flaky.FAIL_FLAKY_TESTS_PROPERTY));
+        // CHECKSTYLE:ON
 
         connection = TestHelper.testConnection();
+
+        TestHelper.disableDbCdc(connection);
+        TestHelper.dropAllTables();
+
         connection.execute("DELETE FROM ASNCDC.IBMSNAP_REGISTER");
         connection.execute(
                 "CREATE TABLE tablea (id int not null, cola varchar(30), primary key (id))",

src/test/java/io/debezium/connector/db2/Db2ReselectColumnsProcessorIT.java

@@ -27,6 +27,13 @@ public class Db2ReselectColumnsProcessorIT extends AbstractReselectProcessorTest
     @Before
     public void beforeEach() throws Exception {
         connection = TestHelper.testConnection();
+
+        TestHelper.disableDbCdc(connection);
+        TestHelper.disableTableCdc(connection, "dbz4321");
+        connection.execute("DROP TABLE IF EXISTS dbz4321");
+
+        connection.execute("DELETE FROM ASNCDC.IBMSNAP_REGISTER");
+
         initializeConnectorTestFramework();
         Testing.Files.delete(TestHelper.DB_HISTORY_PATH);
         super.beforeEach();

src/test/java/io/debezium/connector/db2/IncrementalSnapshotIT.java

@@ -62,18 +62,48 @@ public void before() throws SQLException {
     @After
     public void after() throws SQLException {
         if (connection != null) {
-            TestHelper.disableDbCdc(connection);
-            TestHelper.disableTableCdc(connection, "A");
-            TestHelper.disableTableCdc(connection, "B");
-            TestHelper.disableTableCdc(connection, "DEBEZIUM_SIGNAL");
-            connection.rollback();
-            connection.execute(
-                    "DROP TABLE IF EXISTS a",
-                    "DROP TABLE IF EXISTS b",
-                    "DROP TABLE IF EXISTS debezium_signal");
-            connection.execute("DELETE FROM ASNCDC.IBMSNAP_REGISTER");
-            connection.execute("DELETE FROM ASNCDC.IBMQREP_COLVERSION");
-            connection.execute("DELETE FROM ASNCDC.IBMQREP_TABVERSION");
+            try {
+                TestHelper.disableDbCdc(connection);
+            }
+            catch (Exception e) {
+                e.printStackTrace();
+            }
+            try {
+                TestHelper.disableTableCdc(connection, "A");
+            }
+            catch (Exception e) {
+                e.printStackTrace();
+            }
+            try {
+                TestHelper.disableTableCdc(connection, "B");
+            }
+            catch (Exception e) {
+                e.printStackTrace();
+            }
+            try {
+                TestHelper.disableTableCdc(connection, "DEBEZIUM_SIGNAL");
+            }
+            catch (Exception e) {
+                e.printStackTrace();
+            }
+            // connection.rollback();
+            try {
+                connection.execute(
+                        "DROP TABLE IF EXISTS a",
+                        "DROP TABLE IF EXISTS b",
+                        "DROP TABLE IF EXISTS debezium_signal");
+            }
+            catch (Exception e) {
+                e.printStackTrace();
+            }
+            try {
+                connection.execute("DELETE FROM ASNCDC.IBMSNAP_REGISTER");
+                connection.execute("DELETE FROM ASNCDC.IBMQREP_COLVERSION");
+                connection.execute("DELETE FROM ASNCDC.IBMQREP_TABVERSION");
+            }
+            catch (Exception e) {
+                e.printStackTrace();
+            }
             connection.close();
         }
     }
@@ -194,4 +224,16 @@ public void snapshotWithAdditionalConditionWithRestart() throws Exception {
         super.snapshotWithAdditionalConditionWithRestart();
     }
 
+    @Test
+    @Flaky("DBZ-7478")
+    public void snapshotWithAdditionalCondition() throws Exception {
+        super.snapshotWithAdditionalCondition();
+    }
+
+    @Test
+    @Flaky("DBZ-XXXX")
+    public void snapshotOnlyWithRestart() throws Exception {
+        super.snapshotOnlyWithRestart();
+    }
+
 }

src/test/java/io/debezium/connector/db2/util/TestHelper.java

@@ -125,6 +125,7 @@ public static void enableDbCdc(Db2Connection connection) throws SQLException {
         Statement stmt = connection.connection().createStatement();
         boolean isNotrunning = true;
         int count = 0;
+        int retries = 0;
         while (isNotrunning) {
             ResultSet rs = stmt.executeQuery(STATUS_DB_CDC);
             while (rs.next()) {
@@ -134,14 +135,25 @@ public static void enableDbCdc(Db2Connection connection) throws SQLException {
                 if (test.contains("is doing work")) {
                     isNotrunning = false;
                 }
+                else if (test.contains("ASN0510E")) {
+                    // Per https://www.ibm.com/docs/en/db2/11.5?topic=messages-asn0000-asn0999#asn0510e
+                    // The command was not executed and should be retried in this use case.
+                    LOGGER.debug("ASN0510E detected, command was not processed and requires retry.");
+                    connection.execute(ENABLE_DB_CDC);
+                    retries++;
+                    count = 0;
+                }
                 else {
                     try {
                         Thread.sleep(1000);
                     }
                     catch (InterruptedException e) {
                     }
                 }
-                if (count++ > 30) {
+                if (retries > 5) {
+                    throw new SQLException("Maximum ASNCAP server start requests exceeded");
+                }
+                else if (count++ > 60) {
                     throw new SQLException("ASNCAP server did not start.");
                 }
             }