Resolve conflicts

Author: yuxiqian

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToIcebergE2eITCase.java

@@ -37,6 +37,7 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.BindMode;
@@ -69,15 +70,10 @@
 
 /** End-to-end tests for mysql cdc to Iceberg pipeline job. */
 public class MySqlToIcebergE2eITCase extends PipelineTestEnvironment {
-    private static final Logger LOG = LoggerFactory.getLogger(MySqlToIcebergE2eITCase.class);
 
-    public static final Duration TESTCASE_TIMEOUT = Duration.ofMinutes(3);
+    private static final Logger LOG = LoggerFactory.getLogger(MySqlToIcebergE2eITCase.class);
 
-    // ------------------------------------------------------------------------------------------
-    // MySQL Variables (we always use MySQL as the data source for easier verifying)
-    // ------------------------------------------------------------------------------------------
-    protected static final String MYSQL_TEST_USER = "mysqluser";
-    protected static final String MYSQL_TEST_PASSWORD = "mysqlpw";
+    @TempDir public Path temporaryFolder;
 
     @org.testcontainers.junit.jupiter.Container
     public static final MySqlContainer MYSQL =
@@ -351,7 +347,7 @@ private void validateSinkResult(
             throws InterruptedException {
         runInContainerAsRoot(jobManager, "chmod", "0777", "-R", warehouse);
         LOG.info("Verifying Iceberg {}::{}::{} results...", warehouse, database, table);
-        long deadline = System.currentTimeMillis() + TESTCASE_TIMEOUT.toMillis();
+        long deadline = System.currentTimeMillis() + EVENT_WAITING_TIMEOUT.toMillis();
         List<String> results = Collections.emptyList();
         while (System.currentTimeMillis() < deadline) {
             try {
@@ -364,7 +360,7 @@ private void validateSinkResult(
                 LOG.info(
                         "Successfully verified {} records in {} seconds.",
                         expected.size(),
-                        (System.currentTimeMillis() - deadline + TESTCASE_TIMEOUT.toMillis())
+                        (System.currentTimeMillis() - deadline + EVENT_WAITING_TIMEOUT.toMillis())
                                 / 1000);
                 return;
             } catch (Exception e) {

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlE2eITCase.java

@@ -178,22 +178,11 @@ void testSyncWholeDatabaseInBatchMode() throws Exception {
                         MYSQL_TEST_PASSWORD,
                         mysqlInventoryDatabase.getDatabaseName(),
                         parallelism);
-        Path mysqlCdcJar = TestUtils.getResource("mysql-cdc-pipeline-connector.jar");
-        Path valuesCdcJar = TestUtils.getResource("values-cdc-pipeline-connector.jar");
-        Path mysqlDriverJar = TestUtils.getResource("mysql-driver.jar");
-        submitPipelineJob(pipelineJob, mysqlCdcJar, valuesCdcJar, mysqlDriverJar);
+        submitPipelineJob(pipelineJob);
         waitUntilJobRunning(Duration.ofSeconds(30));
-        LOG.info("Pipeline job is running");
-        waitUntilSpecificEvent(
-                String.format(
-                        "DataChangeEvent{tableId=%s.customers, before=[], after=[104, user_4, Shanghai, 123567891234], op=INSERT, meta=()}",
-                        mysqlInventoryDatabase.getDatabaseName()));
-        waitUntilSpecificEvent(
-                String.format(
-                        "DataChangeEvent{tableId=%s.products, before=[], after=[109, spare tire, 24 inch spare tire, 22.2, null, null, null], op=INSERT, meta=()}",
-                        mysqlInventoryDatabase.getDatabaseName()));
 
         validateResult(
+                dbNameFormatter,
                 "CreateTableEvent{tableId=%s.customers, schema=columns={`id` INT NOT NULL,`name` VARCHAR(255) NOT NULL 'flink',`address` VARCHAR(1024),`phone_number` VARCHAR(512)}, primaryKeys=id, options=()}",
                 "CreateTableEvent{tableId=%s.products, schema=columns={`id` INT NOT NULL,`name` VARCHAR(255) NOT NULL 'flink',`description` VARCHAR(512),`weight` FLOAT,`enum_c` STRING 'red',`json_c` STRING,`point_c` STRING}, primaryKeys=id, options=()}",
                 "DataChangeEvent{tableId=%s.customers, before=[], after=[104, user_4, Shanghai, 123567891234], op=INSERT, meta=()}",
@@ -238,7 +227,6 @@ void testSchemaChangeEvents() throws Exception {
                         parallelism);
         submitPipelineJob(pipelineJob);
         waitUntilJobRunning(Duration.ofSeconds(30));
-        LOG.info("Pipeline job is running");
 
         validateResult(
                 dbNameFormatter,
@@ -258,8 +246,6 @@ void testSchemaChangeEvents() throws Exception {
                 "DataChangeEvent{tableId=%s.products, before=[], after=[101, scooter, Small 2-wheel scooter, 3.14, red, {\"key1\": \"value1\"}, {\"coordinates\":[1,1],\"type\":\"Point\",\"srid\":0}], op=INSERT, meta=()}",
                 "DataChangeEvent{tableId=%s.products, before=[], after=[102, car battery, 12V car battery, 8.1, white, {\"key2\": \"value2\"}, {\"coordinates\":[2,2],\"type\":\"Point\",\"srid\":0}], op=INSERT, meta=()}");
 
-        LOG.info("Begin incremental reading stage.");
-        // generate binlogs
         String mysqlJdbcUrl =
                 String.format(
                         "jdbc:mysql://%s:%s/%s",