Add test back.

Author: lvyanquan

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/test/java/org/apache/flink/cdc/connectors/oracle/source/NewlyAddedTableITCase.java

@@ -19,6 +19,7 @@
 
 import org.apache.flink.cdc.connectors.oracle.testutils.OracleTestUtils.FailoverPhase;
 import org.apache.flink.cdc.connectors.oracle.testutils.OracleTestUtils.FailoverType;
+import org.apache.flink.cdc.connectors.utils.TestValuesHelper;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.StateRecoveryOptions;
 import org.apache.flink.core.execution.JobClient;
@@ -686,7 +687,7 @@ private void testNewlyAddedTableOneByOne(
             }
             fetchedDataList.addAll(expectedSnapshotDataThisRound);
             waitForUpsertSinkSize("sink", fetchedDataList.size());
-            assertEqualsInAnyOrder(fetchedDataList, TestValuesSinkHelper.getUpsertResults("sink"));
+            assertEqualsInAnyOrder(fetchedDataList, TestValuesHelper.getUpsertResults("sink"));
             // Wait 1s until snapshot phase finished, make sure the binlog data is not lost.
             Thread.sleep(1000L);
 
@@ -729,7 +730,7 @@ private void testNewlyAddedTableOneByOne(
             // the result size of sink may arrive fetchedDataList.size() with old data, wait one
             // checkpoint to wait retract old record and send new record
             Thread.sleep(1000);
-            assertEqualsInAnyOrder(fetchedDataList, TestValuesSinkHelper.getUpsertResults("sink"));
+            assertEqualsInAnyOrder(fetchedDataList, TestValuesHelper.getUpsertResults("sink"));
 
             // step 6: trigger savepoint
             if (round != captureAddressTables.length - 1) {

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/test/java/org/apache/flink/cdc/connectors/oracle/testutils/OracleTestUtils.java

@@ -18,7 +18,7 @@
 package org.apache.flink.cdc.connectors.oracle.testutils;
 
 import org.apache.flink.api.common.JobID;
-import org.apache.flink.cdc.connectors.oracle.source.TestValuesSinkHelper;
+import org.apache.flink.cdc.connectors.utils.TestValuesHelper;
 import org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServices;
 import org.apache.flink.runtime.highavailability.nonha.embedded.HaLeadershipControl;
 import org.apache.flink.runtime.minicluster.MiniCluster;
@@ -119,7 +119,7 @@ public static void waitForUpsertSinkSize(String sinkName, int expectedSize)
     public static int upsertSinkSize(String sinkName) {
         synchronized (TestValuesTableFactory.class) {
             try {
-                return TestValuesSinkHelper.getUpsertResults(sinkName).size();
+                return TestValuesHelper.getUpsertResults(sinkName).size();
             } catch (IllegalArgumentException e) {
                 // job is not started yet
                 return 0;

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/NewlyAddedTableITCase.java

@@ -21,6 +21,7 @@
 import org.apache.flink.cdc.connectors.postgres.testutils.PostgresTestUtils;
 import org.apache.flink.cdc.connectors.postgres.testutils.UniqueDatabase;
 import org.apache.flink.cdc.connectors.utils.ExternalResourceProxy;
+import org.apache.flink.cdc.connectors.utils.TestValuesHelper;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.StateRecoveryOptions;
 import org.apache.flink.core.execution.JobClient;
@@ -724,7 +725,7 @@ private void testNewlyAddedTableOneByOne(
             }
             fetchedDataList.addAll(expectedSnapshotDataThisRound);
             PostgresTestUtils.waitForUpsertSinkSize("sink", fetchedDataList.size());
-            assertEqualsInAnyOrder(fetchedDataList, TestValuesSinkHelper.getUpsertResults("sink"));
+            assertEqualsInAnyOrder(fetchedDataList, TestValuesHelper.getUpsertResults("sink"));
             // Wait 1s until snapshot phase finished, make sure the binlog data is not lost.
             Thread.sleep(1000L);
 
@@ -767,7 +768,7 @@ private void testNewlyAddedTableOneByOne(
             // the result size of sink may arrive fetchedDataList.size() with old data, wait one
             // checkpoint to wait retract old record and send new record
             Thread.sleep(1000);
-            assertEqualsInAnyOrder(fetchedDataList, TestValuesSinkHelper.getUpsertResults("sink"));
+            assertEqualsInAnyOrder(fetchedDataList, TestValuesHelper.getUpsertResults("sink"));
 
             // step 6: trigger savepoint
             if (round != captureAddressTables.length - 1) {

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/testutils/PostgresTestUtils.java

@@ -18,7 +18,7 @@
 package org.apache.flink.cdc.connectors.postgres.testutils;
 
 import org.apache.flink.api.common.JobID;
-import org.apache.flink.cdc.connectors.postgres.source.TestValuesSinkHelper;
+import org.apache.flink.cdc.connectors.utils.TestValuesHelper;
 import org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServices;
 import org.apache.flink.runtime.highavailability.nonha.embedded.HaLeadershipControl;
 import org.apache.flink.runtime.minicluster.MiniCluster;
@@ -111,7 +111,7 @@ public static void waitForUpsertSinkSize(String sinkName, int expectedSize)
     public static int upsertSinkSize(String sinkName) {
         synchronized (TestValuesTableFactory.class) {
             try {
-                return TestValuesSinkHelper.getUpsertResults(sinkName).size();
+                return TestValuesHelper.getUpsertResults(sinkName).size();
             } catch (IllegalArgumentException e) {
                 // job is not started yet
                 return 0;