Resolve conflicts

yuxiqian · yuxiqian · commit fbf3d3056c2d · 2025-02-24T19:56:03.000+08:00
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/LegacyMongoDBTestBase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/LegacyMongoDBTestBase.java
@@ -30,7 +30,6 @@
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.Network;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
-import org.testcontainers.junit.jupiter.Testcontainers;
 import org.testcontainers.lifecycle.Startables;
 
 import java.util.stream.Stream;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/polardbx/PolardbxSourceTestBase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/polardbx/PolardbxSourceTestBase.java
@@ -18,8 +18,8 @@
 package org.apache.flink.cdc.connectors.polardbx;
 
 import org.apache.flink.cdc.common.utils.TestCaseUtils;
-import org.apache.flink.cdc.connectors.utils.AbstractTestBaseProxy;
 import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.test.util.AbstractTestBase;
 import org.apache.flink.types.Row;
 
 import org.apache.commons.lang3.StringUtils;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oceanbase-cdc/src/test/java/org/apache/flink/cdc/connectors/oceanbase/OceanBaseTestBase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oceanbase-cdc/src/test/java/org/apache/flink/cdc/connectors/oceanbase/OceanBaseTestBase.java
@@ -18,10 +18,10 @@
 package org.apache.flink.cdc.connectors.oceanbase;
 
 import org.apache.flink.cdc.connectors.oceanbase.testutils.OceanBaseCdcMetadata;
-import org.apache.flink.cdc.connectors.utils.AbstractTestBaseProxy;
 import org.apache.flink.cdc.connectors.utils.StaticExternalResourceProxy;
 import org.apache.flink.table.planner.factories.TestValuesTableFactory;
 import org.apache.flink.table.utils.LegacyRowResource;
+import org.apache.flink.test.util.AbstractTestBase;
 
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.extension.RegisterExtension;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/PostgresTestBase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/PostgresTestBase.java
@@ -20,8 +20,8 @@
 import org.apache.flink.cdc.connectors.postgres.source.PostgresConnectionPoolFactory;
 import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfigFactory;
 import org.apache.flink.cdc.connectors.postgres.testutils.UniqueDatabase;
-import org.apache.flink.cdc.connectors.utils.AbstractTestBaseProxy;
 import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.test.util.AbstractTestBase;
 import org.apache.flink.types.Row;
 
 import io.debezium.config.Configuration;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceExampleTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceExampleTest.java
@@ -44,6 +44,8 @@
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.sql.SQLException;
 import java.util.ArrayList;
@@ -59,6 +61,8 @@
 /** Tests for Postgres Source based on incremental snapshot framework . */
 class PostgresSourceExampleTest extends PostgresTestBase {
 
+    private static final Logger LOG = LoggerFactory.getLogger(PostgresSourceExampleTest.class);
+
     private static final String DB_NAME_PREFIX = "postgres";
     private static final String SCHEMA_NAME = "inventory";
     private static final String TABLE_ID = SCHEMA_NAME + ".products";
@@ -192,7 +196,7 @@ void testConsumingAllEvents() throws Exception {
         List<String> snapshotActualRecords = formatResult(snapshotRowDataList, dataType);
         assertEqualsInAnyOrder(Arrays.asList(snapshotExpectedRecords), snapshotActualRecords);
 
-        log.info("All snapshot data consumed!");
+        LOG.info("All snapshot data consumed!");
 
         // step-2: make 6 change events in one PostgreSQL transaction
         makeWalEvents(getConnection(), TABLE_ID);
@@ -211,14 +215,14 @@ void testConsumingAllEvents() throws Exception {
         List<RowData> walRowDataList = new ArrayList<>();
         for (int i = 0; i < walExpectedRecords.length && iterator.hasNext(); i++) {
             RowData rowData = iterator.next();
-            log.info("step 3: consume wal event: {}", rowData);
+            LOG.info("step 3: consume wal event: {}", rowData);
             walRowDataList.add(rowData);
         }
 
         List<String> walActualRecords = formatResult(walRowDataList, dataType);
         assertEqualsInAnyOrder(Arrays.asList(walExpectedRecords), walActualRecords);
 
-        log.info("All streaming events consumed!");
+        LOG.info("All streaming events consumed!");
 
         // stop the worker
         iterator.close();
@@ -281,7 +285,7 @@ private void waitForReplicationSlotReady(PostgresConnection connection) throws S
         SlotState slotState = connection.getReplicationSlotState(SLOT_NAME, PLUGIN_NAME);
 
         while (slotState == null) {
-            log.info("Waiting until the replication slot is ready ...");
+            LOG.info("Waiting until the replication slot is ready ...");
             try {
                 Thread.sleep(2000L);
             } catch (InterruptedException e) {
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/SqlServerTestBase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/SqlServerTestBase.java
@@ -17,11 +17,10 @@
 
 package org.apache.flink.cdc.connectors.sqlserver;
 
-import org.apache.flink.cdc.connectors.utils.AbstractTestBaseProxy;
 import org.apache.flink.cdc.connectors.sqlserver.source.SqlServerSourceTestBase;
 import org.apache.flink.cdc.connectors.sqlserver.source.config.SqlServerSourceConfigFactory;
 import org.apache.flink.table.planner.factories.TestValuesTableFactory;
-import org.apache.flink.test.util.AbstractTestBaseJUnit4;
+import org.apache.flink.test.util.AbstractTestBase;
 
 import org.assertj.core.api.Assertions;
 import org.awaitility.Awaitility;
@@ -50,7 +49,6 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import static org.junit.Assert.assertNotNull;
 import static org.testcontainers.containers.MSSQLServerContainer.MS_SQL_SERVER_PORT;
 
 /** Utility class for sqlserver tests. */
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/reader/SqlserverSourceReaderTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/reader/SqlserverSourceReaderTest.java
@@ -54,8 +54,9 @@
 import io.debezium.relational.TableId;
 import io.debezium.relational.history.TableChanges;
 import org.apache.kafka.connect.source.SourceRecord;
+import org.assertj.core.api.Assertions;
 import org.awaitility.Awaitility;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -64,7 +65,6 @@
 import java.sql.Statement;
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
@@ -73,13 +73,12 @@
 
 import static org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit.STREAM_SPLIT_ID;
 import static org.apache.flink.core.io.InputStatus.MORE_AVAILABLE;
-import static org.junit.Assert.assertEquals;
 
 /** Tests for Sqlserver incremental source reader. */
-public class SqlserverSourceReaderTest extends SqlServerSourceTestBase {
+class SqlserverSourceReaderTest extends SqlServerSourceTestBase {
 
     @Test
-    public void testIncrementalReadFailoverCrossTransaction() throws Exception {
+    void testIncrementalReadFailoverCrossTransaction() throws Exception {
 
         String databaseName = "customer";
         String tableName = "dbo.customers";
@@ -141,11 +140,11 @@ public void testIncrementalReadFailoverCrossTransaction() throws Exception {
                 };
         // the 2 records are produced by 1 operations
         List<String> actualRecords = consumeRecords(reader, dataType);
-        assertEqualsInOrder(Arrays.asList(expectedRecords), actualRecords);
+        Assertions.assertThat(actualRecords).containsExactlyInAnyOrder(expectedRecords);
         // check the binlog split state
         List<SourceSplitBase> splitsState = reader.snapshotState(1L);
 
-        assertEquals(1, splitsState.size());
+        Assertions.assertThat(splitsState).hasSize(1);
         reader.close();
 
         // step-3: mock failover from a restored state
@@ -166,7 +165,7 @@ public void testIncrementalReadFailoverCrossTransaction() throws Exception {
 
         // the 4 records are produced by 3 operations
         List<String> restRecords = consumeRecords(restartReader, dataType);
-        assertEqualsInOrder(Arrays.asList(expectedRestRecords), restRecords);
+        Assertions.assertThat(restRecords).containsExactlyInAnyOrder(expectedRestRecords);
         restartReader.close();
     }
 
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/reader/fetch/SqlServerScanFetchTaskTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/reader/fetch/SqlServerScanFetchTaskTest.java
@@ -55,7 +55,6 @@
 
 import static org.apache.flink.cdc.connectors.sqlserver.SqlServerTestBase.getConfigFactory;
 import static org.apache.flink.cdc.connectors.sqlserver.source.utils.SqlServerConnectionUtils.createSqlServerConnection;
-import static org.testcontainers.containers.MSSQLServerContainer.MS_SQL_SERVER_PORT;
 
 /**
  * Tests for {@link
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-test-util/src/main/java/org/apache/flink/cdc/connectors/utils/AbstractTestBaseProxy.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-test-util/src/main/java/org/apache/flink/cdc/connectors/utils/AbstractTestBaseProxy.java
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/test/java/org/apache/flink/cdc/connectors/tidb/TiDBTestBase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/test/java/org/apache/flink/cdc/connectors/tidb/TiDBTestBase.java
@@ -17,6 +17,8 @@
 
 package org.apache.flink.cdc.connectors.tidb;
 
+import org.apache.flink.test.util.AbstractTestBase;
+
 import com.alibaba.dcm.DnsCacheManipulator;
 import org.apache.commons.lang3.RandomUtils;
 import org.assertj.core.api.Assertions;
@@ -31,7 +33,6 @@
 import org.testcontainers.containers.Network;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
 import org.testcontainers.junit.jupiter.Container;
-import org.testcontainers.junit.jupiter.Testcontainers;
 import org.testcontainers.lifecycle.Startables;
 
 import java.net.URL;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-vitess-cdc/src/test/java/org/apache/flink/cdc/connectors/vitess/VitessTestBase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-vitess-cdc/src/test/java/org/apache/flink/cdc/connectors/vitess/VitessTestBase.java
@@ -17,8 +17,8 @@
 
 package org.apache.flink.cdc.connectors.vitess;
 
-import org.apache.flink.cdc.connectors.utils.AbstractTestBaseProxy;
 import org.apache.flink.cdc.connectors.vitess.container.VitessContainer;
+import org.apache.flink.test.util.AbstractTestBase;
 
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.AfterAll;
diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/SerializerTestBase.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/SerializerTestBase.java
@@ -497,8 +497,7 @@ void testDuplicate() throws Exception {
     private void deepEquals(String message, T should, T is) {
         Assertions.assertThat(is)
                 .withFailMessage(message)
-                .satisfies(
-                        CustomEqualityMatcher.deeplyEquals(should).withChecker(checker)::matches);
+                .matches(CustomEqualityMatcher.deeplyEquals(should).withChecker(checker));
     }
 
     // --------------------------------------------------------------------------------------------
@@ -596,10 +595,9 @@ public void run() {
                         Assertions.assertThat(copySerdeTestItem)
                                 .withFailMessage(
                                         "Serialization/Deserialization cycle resulted in an object that are not equal to the original.")
-                                .satisfies(
+                                .matches(
                                         CustomEqualityMatcher.deeplyEquals(testItem)
-                                                        .withChecker(checker)
-                                                ::matches);
+                                                .withChecker(checker));
 
                         // try to enforce some upper bound to the test time
                         if (System.nanoTime() >= endTimeNanos) {
