[hotfix][mysql] Fix failed test LegacyMySqlSourceTest#testSnapshotOnF…

…ailedSource (apache#2384) * [hotfix][mysql] fix failed test LegacyMySqlSourceTest#testSnapshotOnFailedSource * [hotfix][mysql] fix failed test LegacyMySqlSourceTest#testSnapshotOnFailedSource (cherry picked from commit b8f468e)
zhongqishang · Dec 7, 2023 · 22da1e2 · 22da1e2
1 parent 0717756
commit 22da1e2
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 7 deletions.
diff --git a/flink-connector-debezium/src/main/java/com/ververica/cdc/debezium/internal/Handover.java b/flink-connector-debezium/src/main/java/com/ververica/cdc/debezium/internal/Handover.java
@@ -17,6 +17,7 @@
 package com.ververica.cdc.debezium.internal;
 
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.util.ExceptionUtils;
 
 import io.debezium.engine.ChangeEvent;
@@ -223,4 +224,9 @@ public static boolean isGentlyClosedException(Throwable cause) {
                     && GENTLY_CLOSED_MESSAGE.equals(((ClosedException) cause).getMessage());
         }
     }
+
+    @VisibleForTesting
+    public Object getLock() {
+        return lock;
+    }
 }
diff --git a/...tor-mysql-cdc/src/test/java/com/ververica/cdc/connectors/mysql/LegacyMySqlSourceTest.java b/...tor-mysql-cdc/src/test/java/com/ververica/cdc/connectors/mysql/LegacyMySqlSourceTest.java
@@ -1028,15 +1028,22 @@ public void go() throws Exception {
                 assertTrue(historyState.list.size() > 0);
                 assertTrue(offsetState.list.size() > 0);
 
-                // Step-3: mock the engine stop due to underlying debezium exception, trigger a
-                // checkpoint on failed source
                 final Handover handover = source.getHandover();
-                handover.reportError(new DebeziumException("Mocked debezium exception"));
-                handover.close();
                 try {
-                    synchronized (sourceContext.getCheckpointLock()) {
-                        // trigger checkpoint-2
-                        source.snapshotState(new StateSnapshotContextSynchronousImpl(102, 102));
+                    // Here need handover's lock to make sure the error will not be sent by method
+                    // pollNext()
+                    // before the method snapshotState() is invoked
+                    synchronized (handover.getLock()) {
+                        // Step-3: mock the engine stop due to underlying debezium exception,
+                        // trigger a
+                        // checkpoint on failed source
+                        handover.reportError(new DebeziumException("Mocked debezium exception"));
+                        handover.close();
+                        synchronized (sourceContext.getCheckpointLock()) {
+                            // trigger checkpoint-2
+                            source.snapshotState(new StateSnapshotContextSynchronousImpl(102, 102));
+                        }
+                        handover.getLock().notifyAll();
                     }
                     fail("Should fail.");
                 } catch (Exception e) {