[hotfix][tests] Fix MySQL ITCase failure in non-UTC timezone

Author: yuxiqian

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

@@ -21,7 +21,6 @@
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.cdc.common.utils.TestCaseUtils;
 import org.apache.flink.cdc.connectors.mysql.debezium.DebeziumUtils;
-import org.apache.flink.cdc.connectors.mysql.testutils.MySqlContainer;
 import org.apache.flink.cdc.connectors.mysql.testutils.TestTable;
 import org.apache.flink.cdc.connectors.mysql.testutils.TestTableSchemas;
 import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
@@ -38,20 +37,15 @@
 import org.apache.flink.streaming.api.operators.collect.CollectSinkOperator;
 import org.apache.flink.streaming.api.operators.collect.CollectSinkOperatorFactory;
 import org.apache.flink.streaming.api.operators.collect.CollectStreamSink;
-import org.apache.flink.test.junit5.MiniClusterExtension;
 import org.apache.flink.util.FlinkRuntimeException;
 
 import io.debezium.connector.mysql.MySqlConnection;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.testcontainers.containers.output.Slf4jLogConsumer;
 
 import java.io.File;
 import java.lang.reflect.Field;
@@ -70,29 +64,18 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Properties;
+import java.util.Random;
 import java.util.UUID;
 
+import static org.apache.flink.cdc.connectors.mysql.source.MySqlSourceTestBase.DEFAULT_PARALLELISM;
+
 /**
  * Integration tests for handling schema changes regard to renaming multiple tables within a single
  * statement.
  */
-public class MySqlMultipleTablesRenamingITCase {
+public class MySqlMultipleTablesRenamingITCase extends MySqlSourceTestBase {
     private static final Logger LOG =
             LoggerFactory.getLogger(MySqlMultipleTablesRenamingITCase.class);
-    @RegisterExtension static MiniClusterExtension miniCluster = new MiniClusterExtension();
-
-    @SuppressWarnings("unchecked")
-    private static final MySqlContainer MYSQL_CONTAINER =
-            (MySqlContainer)
-                    new MySqlContainer()
-                            .withConfigurationOverride(
-                                    buildMySqlConfigWithTimezone(
-                                            getResourceFolder(), getSystemTimeZone()))
-                            .withSetupSQL("docker/setup.sql")
-                            .withDatabaseName("flink-test")
-                            .withUsername("flinkuser")
-                            .withPassword("flinkpw")
-                            .withLogConsumer(new Slf4jLogConsumer(LOG));
 
     private final UniqueDatabase customDatabase =
             new UniqueDatabase(MYSQL_CONTAINER, "customer", "mysqluser", "mysqlpw");
@@ -101,25 +84,15 @@ public class MySqlMultipleTablesRenamingITCase {
 
     private MySqlConnection connection;
 
-    @BeforeAll
-    public static void before() throws Exception {
-        MYSQL_CONTAINER.start();
-    }
-
-    @AfterAll
-    public static void after() throws Exception {
-        MYSQL_CONTAINER.stop();
-    }
-
-    @BeforeEach
-    void prepare() throws Exception {
+    @Before
+    public void prepare() throws Exception {
         connection = getConnection();
         customDatabase.createAndInitialize();
         flushLogs();
     }
 
-    @AfterEach
-    void tearDown() throws Exception {
+    @After
+    public void tearDown() throws Exception {
         customDatabase.dropDatabase();
         connection.close();
     }
@@ -146,7 +119,7 @@ void tearDown() throws Exception {
      * during schema updates.
      */
     @Test
-    void testRenameTablesWithinSingleStatement() throws Exception {
+    public void testRenameTablesWithinSingleStatement() throws Exception {
         // Build Flink job
         StreamExecutionEnvironment env = getExecutionEnvironment();
         MySqlSource<String> source = getSourceBuilder().build();
@@ -269,6 +242,8 @@ private MySqlSourceBuilder<String> getSourceBuilder() {
                 .password(customDatabase.getPassword())
                 .databaseList(customDatabase.getDatabaseName())
                 .tableList(customers.getTableId())
+                .serverId(getServerId())
+                .serverTimeZone("UTC")
                 .deserializer(new JsonDebeziumDeserializationSchema());
     }
 
@@ -392,4 +367,10 @@ private StreamExecutionEnvironment getExecutionEnvironment() {
         env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
         return env;
     }
+
+    private String getServerId() {
+        final Random random = new Random();
+        int serverId = random.nextInt(100) + 5400;
+        return serverId + "-" + (serverId + DEFAULT_PARALLELISM);
+    }
 }

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

@@ -714,6 +714,7 @@ public void testSourceMetrics() throws Exception {
                         .password(customDatabase.getPassword())
                         .deserializer(new StringDebeziumDeserializationSchema())
                         .serverId(getServerId())
+                        .serverTimeZone("UTC")
                         .build();
         DataStreamSource<String> stream =
                 env.fromSource(source, WatermarkStrategy.noWatermarks(), "MySQL CDC Source");