[hotfix][tests] Fix unstable OceanBaseMySQLModelITCase (#3831)

Signed-off-by: yuxiqian <34335406+yuxiqian@users.noreply.github.com>
apache · Jan 3, 2025 · ddb5f00 · ddb5f00
1 parent 7025537
commit ddb5f00
Showing 1 changed file with 20 additions and 1 deletion.
diff --git a/...c/test/java/org/apache/flink/cdc/connectors/oceanbase/table/OceanBaseMySQLModeITCase.java b/...c/test/java/org/apache/flink/cdc/connectors/oceanbase/table/OceanBaseMySQLModeITCase.java
@@ -104,10 +104,24 @@ protected String logProxyOptionsString() {
                 + String.format(" 'rootserver-list' = '%s'", METADATA.getRsList());
     }
 
+    /**
+     * Current OceanBase connector uses timestamp (in seconds) to mark the offset during the
+     * transition from {@code SNAPSHOT} to {@code STREAMING} mode. Thus, if some snapshot inserting
+     * events are too close to the transitioning offset, snapshot inserting events might be emitted
+     * multiple times. <br>
+     * This could be safely removed after switching to incremental snapshot framework which provides
+     * Exactly-once guarantee.
+     */
+    private void waitForTableInitialization() throws InterruptedException {
+        Thread.sleep(5000L);
+    }
+
     @Test
     public void testTableList() throws Exception {
         inventoryDatabase = new UniqueDatabase(OB_SERVER, "inventory");
         inventoryDatabase.createAndInitialize("mysql");
+        waitForTableInitialization();
+
         String sourceDDL =
                 String.format(
                         "CREATE TABLE ob_source ("
@@ -212,6 +226,8 @@ public void testTableList() throws Exception {
     public void testMetadataColumns() throws Exception {
         inventoryDatabase = new UniqueDatabase(OB_SERVER, "inventory");
         inventoryDatabase.createAndInitialize("mysql");
+        waitForTableInitialization();
+
         String sourceDDL =
                 String.format(
                         "CREATE TABLE ob_source ("
@@ -297,6 +313,7 @@ public void testAllDataTypes() throws Exception {
 
         columnTypesDatabase = new UniqueDatabase(OB_SERVER, "column_type_test");
         columnTypesDatabase.createAndInitialize("mysql");
+        waitForTableInitialization();
 
         String sourceDDL =
                 String.format(
@@ -488,6 +505,7 @@ public void testTimeDataTypes(String serverTimeZone) throws Exception {
 
         columnTypesDatabase = new UniqueDatabase(OB_SERVER, "column_type_test");
         columnTypesDatabase.createAndInitialize("mysql");
+        waitForTableInitialization();
 
         String sourceDDL =
                 String.format(
@@ -559,6 +577,7 @@ public void testTimeDataTypes(String serverTimeZone) throws Exception {
     public void testSnapshotOnly() throws Exception {
         inventoryDatabase = new UniqueDatabase(OB_SERVER, "inventory");
         inventoryDatabase.createAndInitialize("mysql");
+        waitForTableInitialization();
 
         String sourceDDL =
                 String.format(
@@ -611,7 +630,7 @@ public void testSnapshotOnly() throws Exception {
 
         while (result.getJobClient().get().getJobStatus().get().equals(JobStatus.RUNNING)) {
             Thread.sleep(100);
-            // Waiting for job to quit, in case if
+            // Waiting for job to finish (SNAPSHOT job will end spontaneously)
         }
     }
 }