fix review

Author: ruanhang1993

flink-connector-postgres-cdc/src/test/java/com/ververica/cdc/connectors/postgres/PostgresTestBase.java

@@ -16,6 +16,7 @@
 
 package com.ververica.cdc.connectors.postgres;
 
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
 import org.apache.flink.test.util.AbstractTestBase;
 
 import io.debezium.config.Configuration;
@@ -137,4 +138,40 @@ protected PostgresConnection createConnection(Map<String, String> properties) {
         Configuration config = Configuration.from(properties);
         return new PostgresConnection(JdbcConfiguration.adapt(config), "test-connection");
     }
+
+    protected void waitForSnapshotStarted(String sinkName) throws InterruptedException {
+        while (sinkSize(sinkName) == 0) {
+            Thread.sleep(300);
+        }
+    }
+
+    protected void waitForSinkResult(String sinkName, List<String> expected)
+            throws InterruptedException {
+        List<String> actual = TestValuesTableFactory.getResults(sinkName);
+        actual = actual.stream().sorted().collect(Collectors.toList());
+        while (actual.size() != expected.size() || !actual.equals(expected)) {
+            actual =
+                    TestValuesTableFactory.getResults(sinkName).stream()
+                            .sorted()
+                            .collect(Collectors.toList());
+            Thread.sleep(1000);
+        }
+    }
+
+    protected void waitForSinkSize(String sinkName, int expectedSize) throws InterruptedException {
+        while (sinkSize(sinkName) < expectedSize) {
+            Thread.sleep(100);
+        }
+    }
+
+    protected int sinkSize(String sinkName) {
+        synchronized (TestValuesTableFactory.class) {
+            try {
+                return TestValuesTableFactory.getRawResults(sinkName).size();
+            } catch (IllegalArgumentException e) {
+                // job is not started yet
+                return 0;
+            }
+        }
+    }
 }

flink-connector-postgres-cdc/src/test/java/com/ververica/cdc/connectors/postgres/table/PostgreSQLConnectorITCase.java

@@ -674,28 +674,4 @@ public void testUpsertMode() throws Exception {
 
         result.getJobClient().get().cancel().get();
     }
-
-    private static void waitForSnapshotStarted(String sinkName) throws InterruptedException {
-        while (sinkSize(sinkName) == 0) {
-            Thread.sleep(300);
-        }
-    }
-
-    private static void waitForSinkSize(String sinkName, int expectedSize)
-            throws InterruptedException {
-        while (sinkSize(sinkName) < expectedSize) {
-            Thread.sleep(100);
-        }
-    }
-
-    private static int sinkSize(String sinkName) {
-        synchronized (TestValuesTableFactory.class) {
-            try {
-                return TestValuesTableFactory.getRawResults(sinkName).size();
-            } catch (IllegalArgumentException e) {
-                // job is not started yet
-                return 0;
-            }
-        }
-    }
 }

flink-connector-postgres-cdc/src/test/java/com/ververica/cdc/connectors/postgres/table/PostgreSQLSavepointITCase.java

@@ -39,7 +39,6 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
-import java.util.stream.Collectors;
 
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.junit.Assert.assertThat;
@@ -228,35 +227,4 @@ private String triggerSavepointWithRetry(JobClient jobClient, String savepointDi
         }
         return null;
     }
-
-    private static void waitForSinkResult(String sinkName, List<String> expected)
-            throws InterruptedException {
-        List<String> actual = TestValuesTableFactory.getResults(sinkName);
-        actual = actual.stream().sorted().collect(Collectors.toList());
-        while (actual.size() != expected.size() || !actual.equals(expected)) {
-            actual =
-                    TestValuesTableFactory.getResults(sinkName).stream()
-                            .sorted()
-                            .collect(Collectors.toList());
-            Thread.sleep(1000);
-        }
-    }
-
-    private static void waitForSinkSize(String sinkName, int expectedSize)
-            throws InterruptedException {
-        while (sinkSize(sinkName) < expectedSize) {
-            Thread.sleep(100);
-        }
-    }
-
-    private static int sinkSize(String sinkName) {
-        synchronized (TestValuesTableFactory.class) {
-            try {
-                return TestValuesTableFactory.getRawResults(sinkName).size();
-            } catch (IllegalArgumentException e) {
-                // job is not started yet
-                return 0;
-            }
-        }
-    }
 }