fix: polardbx test case

Author: yuxiqian

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/LegacyMySqlSourceITCase.java

@@ -49,8 +49,6 @@
 import java.util.Map;
 import java.util.Objects;
 
-import static org.junit.Assert.assertTrue;
-
 /** Integration tests for the legacy {@link MySqlSource}. */
 public class LegacyMySqlSourceITCase extends LegacyMySqlTestBase {
 
@@ -113,9 +111,8 @@ private void testConsumingAllEventsWithJsonFormat(
         // check the snapshot result
         CloseableIterator<Row> snapshot = result.collect();
         waitForSnapshotStarted(snapshot);
-        assertTrue(
-                dataInJsonIsEquals(
-                        fetchRows(snapshot, 1).get(0).toString(), expectSnapshot.toString()));
+
+        assertJsonEquals(extractJsonBody(snapshot.next()), expectSnapshot);
         try (Connection connection = fullTypesDatabase.getJdbcConnection();
                 Statement statement = connection.createStatement()) {
             statement.execute(
@@ -124,10 +121,8 @@ private void testConsumingAllEventsWithJsonFormat(
 
         // check the binlog result
         CloseableIterator<Row> binlog = result.collect();
-        JSONObject expectBinlog = expected.getJSONObject("expected_binlog");
-        assertTrue(
-                dataInJsonIsEquals(
-                        fetchRows(binlog, 1).get(0).toString(), expectBinlog.toString()));
+        JsonNode expectBinlog = expected.get("expected_binlog");
+        assertJsonEquals(extractJsonBody(binlog.next()), expectBinlog);
         result.getJobClient().get().cancel().get();
     }
 

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/polardbx/PolardbxCharsetITCase.java

@@ -167,8 +167,8 @@ public void testCharset() throws Exception {
                                 + " 'scan.incremental.snapshot.chunk.size' = '%s'"
                                 + ")",
                         testName,
-                        HOST_NAME,
-                        PORT,
+                        getHost(),
+                        getPort(),
                         USER_NAME,
                         PASSWORD,
                         DATABASE,

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/polardbx/PolardbxSourceITCase.java

@@ -83,8 +83,8 @@ public void testSingleKey() throws Exception {
                                 + " 'server-time-zone' = 'UTC',"
                                 + " 'server-id' = '%s'"
                                 + ")",
-                        HOST_NAME,
-                        PORT,
+                        getHost(),
+                        getPort(),
                         USER_NAME,
                         PASSWORD,
                         DATABASE,
@@ -234,8 +234,8 @@ public void testFullTypesDdl() throws Exception {
                                 + " 'server-time-zone' = 'UTC',"
                                 + " 'server-id' = '%s'"
                                 + ")",
-                        HOST_NAME,
-                        PORT,
+                        getHost(),
+                        getPort(),
                         USER_NAME,
                         PASSWORD,
                         DATABASE,
@@ -301,8 +301,8 @@ public void testMultiKeys() throws Exception {
                                 + " 'server-time-zone' = 'UTC',"
                                 + " 'server-id' = '%s'"
                                 + ")",
-                        HOST_NAME,
-                        PORT,
+                        getHost(),
+                        getPort(),
                         USER_NAME,
                         PASSWORD,
                         DATABASE,

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/polardbx/PolardbxSourceTestBase.java

@@ -17,13 +17,11 @@
 
 package org.apache.flink.cdc.connectors.polardbx;
 
+import org.apache.flink.cdc.common.utils.TestCaseUtils;
 import org.apache.flink.table.planner.factories.TestValuesTableFactory;
 import org.apache.flink.test.util.AbstractTestBase;
 import org.apache.flink.types.Row;
 
-import com.github.dockerjava.api.model.ExposedPort;
-import com.github.dockerjava.api.model.PortBinding;
-import com.github.dockerjava.api.model.Ports;
 import org.apache.commons.lang3.StringUtils;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -64,36 +62,37 @@
 public abstract class PolardbxSourceTestBase extends AbstractTestBase {
     private static final Logger LOG = LoggerFactory.getLogger(PolardbxSourceTestBase.class);
     private static final Pattern COMMENT_PATTERN = Pattern.compile("^(.*)--.*$");
-    protected static final Integer PORT = 8527;
-    protected static final String HOST_NAME = "127.0.0.1";
-    protected static final String USER_NAME = "polardbx_root";
-    protected static final String PASSWORD = "123456";
+
     private static final String IMAGE_VERSION = "2.1.0";
     private static final DockerImageName POLARDBX_IMAGE =
             DockerImageName.parse("polardbx/polardb-x:" + IMAGE_VERSION);
 
+    protected static final Integer INNER_PORT = 8527;
+    protected static final String USER_NAME = "polardbx_root";
+    protected static final String PASSWORD = "123456";
+    protected static final Duration WAITING_TIMEOUT = Duration.ofMinutes(1);
+
     protected static final GenericContainer POLARDBX_CONTAINER =
             new GenericContainer<>(POLARDBX_IMAGE)
-                    .withExposedPorts(PORT)
+                    .withExposedPorts(INNER_PORT)
                     .withLogConsumer(new Slf4jLogConsumer(LOG))
-                    .withStartupTimeout(Duration.ofMinutes(3))
-                    .withCreateContainerCmdModifier(
-                            c ->
-                                    c.withPortBindings(
-                                            new PortBinding(
-                                                    Ports.Binding.bindPort(PORT),
-                                                    new ExposedPort(PORT))));
+                    .withStartupTimeout(Duration.ofMinutes(3));
+
+    protected static String getHost() {
+        return POLARDBX_CONTAINER.getHost();
+    }
+
+    protected static int getPort() {
+        return POLARDBX_CONTAINER.getMappedPort(INNER_PORT);
+    }
 
     @BeforeClass
-    public static void startContainers() throws InterruptedException {
-        // no need to start container when the port 8527 is listening
-        if (!checkConnection()) {
-            LOG.info("Polardbx connection is not valid, so try to start containers...");
-            Startables.deepStart(Stream.of(POLARDBX_CONTAINER)).join();
-            LOG.info("Containers are started.");
-            // here should wait 10s that make sure the polardbx is ready
-            Thread.sleep(10 * 1000);
-        }
+    public static void startContainers() {
+        Startables.deepStart(Stream.of(POLARDBX_CONTAINER)).join();
+        LOG.info("Containers are started.");
+
+        TestCaseUtils.repeatedCheck(
+                PolardbxSourceTestBase::checkConnection, WAITING_TIMEOUT, Duration.ofSeconds(1));
     }
 
     @AfterClass
@@ -104,7 +103,7 @@ public static void stopContainers() {
     }
 
     protected static String getJdbcUrl() {
-        return String.format("jdbc:mysql://%s:%s", HOST_NAME, PORT);
+        return String.format("jdbc:mysql://%s:%s", getHost(), getPort());
     }
 
     protected static Connection getJdbcConnection() throws SQLException {