DBZ-5525 Fix test failures - EventProcessingFailureHandlingIT

debezium · Aug 18, 2022 · 6f5e858 · 6f5e858
1 parent fcd2664
commit 6f5e858
Showing 1 changed file with 23 additions and 5 deletions.
diff --git a/...erver/src/test/java/io/debezium/connector/sqlserver/EventProcessingFailureHandlingIT.java b/...erver/src/test/java/io/debezium/connector/sqlserver/EventProcessingFailureHandlingIT.java
@@ -69,7 +69,13 @@ public void warn() throws Exception {
 
         start(SqlServerConnector.class, config);
         assertConnectorIsRunning();
-        TestHelper.waitForSnapshotToBeCompleted();
+        TestHelper.waitForStreamingStarted();
+
+        connection.execute("INSERT INTO tablea VALUES (1, 'seed')");
+
+        SourceRecords records = consumeRecordsByTopic(1);
+        Assertions.assertThat(records.recordsForTopic("server1.testDB1.dbo.tablea")).hasSize(1);
+        Assertions.assertThat(records.recordsForTopic("server1.testDB1.dbo.tableb")).isNull();
 
         // Will allow insertion of strings into what was originally a BIGINT NOT NULL column
         // This will cause NumberFormatExceptions which return nulls and thus an error due to the column being NOT NULL
@@ -83,7 +89,7 @@ public void warn() throws Exception {
                     "INSERT INTO tableb VALUES(" + id + ", 'b')");
         }
 
-        SourceRecords records = consumeRecordsByTopic(RECORDS_PER_TABLE);
+        records = consumeRecordsByTopic(RECORDS_PER_TABLE);
         Assertions.assertThat(records.recordsForTopic("server1.testDB1.dbo.tablea")).hasSize(RECORDS_PER_TABLE);
         Assertions.assertThat(records.recordsForTopic("server1.testDB1.dbo.tableb")).isNull();
 
@@ -105,7 +111,13 @@ public void ignore() throws Exception {
 
         start(SqlServerConnector.class, config);
         assertConnectorIsRunning();
-        TestHelper.waitForSnapshotToBeCompleted();
+        TestHelper.waitForStreamingStarted();
+
+        connection.execute("INSERT INTO tablea VALUES (1, 'seed')");
+
+        SourceRecords records = consumeRecordsByTopic(1);
+        Assertions.assertThat(records.recordsForTopic("server1.testDB1.dbo.tablea")).hasSize(1);
+        Assertions.assertThat(records.recordsForTopic("server1.testDB1.dbo.tableb")).isNull();
 
         // Will allow insertion of strings into what was originally a BIGINT NOT NULL column
         // This will cause NumberFormatExceptions which return nulls and thus an error due to the column being NOT NULL
@@ -119,7 +131,7 @@ public void ignore() throws Exception {
                     "INSERT INTO tableb VALUES(" + id + ", 'b')");
         }
 
-        SourceRecords records = consumeRecordsByTopic(RECORDS_PER_TABLE);
+        records = consumeRecordsByTopic(RECORDS_PER_TABLE);
         Assertions.assertThat(records.recordsForTopic("server1.testDB1.dbo.tablea")).hasSize(RECORDS_PER_TABLE);
         Assertions.assertThat(records.recordsForTopic("server1.testDB1.dbo.tableb")).isNull();
     }
@@ -135,7 +147,13 @@ public void fail() throws Exception {
 
         start(SqlServerConnector.class, config);
         assertConnectorIsRunning();
-        TestHelper.waitForSnapshotToBeCompleted();
+        TestHelper.waitForStreamingStarted();
+
+        connection.execute("INSERT INTO tablea VALUES (1, 'seed')");
+
+        SourceRecords records = consumeRecordsByTopic(1);
+        Assertions.assertThat(records.recordsForTopic("server1.testDB1.dbo.tablea")).hasSize(1);
+        Assertions.assertThat(records.recordsForTopic("server1.testDB1.dbo.tableb")).isNull();
 
         // Will allow insertion of strings into what was originally a BIGINT NOT NULL column
         // This will cause NumberFormatExceptions which return nulls and thus an error due to the column being NOT NULL