[FLINK-37152][build] Update Flink version to 1.20

flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/CliFrontendTest.java

@@ -20,7 +20,7 @@
 import org.apache.flink.cdc.composer.PipelineComposer;
 import org.apache.flink.cdc.composer.PipelineExecution;
 import org.apache.flink.cdc.composer.definition.PipelineDef;
-import org.apache.flink.runtime.jobgraph.RestoreMode;
+import org.apache.flink.core.execution.RestoreMode;
 
 import org.apache.flink.shaded.guava31.com.google.common.io.Resources;
 

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/pom.xml

@@ -28,7 +28,8 @@ limitations under the License.
     <artifactId>flink-cdc-pipeline-connector-starrocks</artifactId>
 
     <properties>
-        <starrocks.connector.version>1.2.10_flink-${flink.major.version}</starrocks.connector.version>
+        <!-- TODO: Update this, when StarRocks releases a 1.20 compatible connector. -->
+        <starrocks.connector.version>1.2.10_flink-1.19</starrocks.connector.version>
     </properties>
 
     <dependencies>
@@ -108,4 +109,4 @@ limitations under the License.
             </plugin>
         </plugins>
     </build>
-</project>
+</project>

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/LegacyMongoDBTestBase.java

@@ -17,7 +17,7 @@
 
 package org.apache.flink.cdc.connectors.mongodb;
 
-import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.test.util.AbstractTestBaseJUnit4;
 
 import com.mongodb.ConnectionString;
 import com.mongodb.MongoClientSettings;
@@ -41,7 +41,7 @@
  * Basic class for testing MongoDB source, this contains a MongoDB container which enables change
  * streams.
  */
-public class LegacyMongoDBTestBase extends AbstractTestBase {
+public class LegacyMongoDBTestBase extends AbstractTestBaseJUnit4 {
 
     private static final Logger LOG = LoggerFactory.getLogger(LegacyMongoDBTestBase.class);
 

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/log4j2-test.properties

@@ -15,7 +15,7 @@
 
 # Set root logger level to ERROR to not flood build logs
 # set manually to INFO for debugging purposes
-rootLogger.level=ERROR
+rootLogger.level=DEBUG
 rootLogger.appenderRef.test.ref = TestLogger
 
 appender.testlogger.name = TestLogger

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

@@ -18,7 +18,7 @@
 package org.apache.flink.cdc.connectors.mysql;
 
 import org.apache.flink.cdc.connectors.mysql.testutils.MySqlContainer;
-import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.test.util.AbstractTestBaseJUnit4;
 
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -33,7 +33,7 @@
  * Basic class for testing MySQL binlog source, this contains a MySQL container which enables
  * binlog.
  */
-public abstract class LegacyMySqlTestBase extends AbstractTestBase {
+public abstract class LegacyMySqlTestBase extends AbstractTestBaseJUnit4 {
 
     private static final Logger LOG = LoggerFactory.getLogger(LegacyMySqlTestBase.class);
 

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

@@ -18,7 +18,7 @@
 package org.apache.flink.cdc.connectors.polardbx;
 
 import org.apache.flink.table.planner.factories.TestValuesTableFactory;
-import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.test.util.AbstractTestBaseJUnit4;
 import org.apache.flink.types.Row;
 
 import com.github.dockerjava.api.model.ExposedPort;
@@ -61,7 +61,7 @@
 import static org.junit.Assert.assertTrue;
 
 /** Basic class for testing Database Polardbx which supported the mysql protocol. */
-public abstract class PolardbxSourceTestBase extends AbstractTestBase {
+public abstract class PolardbxSourceTestBase extends AbstractTestBaseJUnit4 {
     private static final Logger LOG = LoggerFactory.getLogger(PolardbxSourceTestBase.class);
     private static final Pattern COMMENT_PATTERN = Pattern.compile("^(.*)--.*$");
     protected static final Integer PORT = 8527;

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oceanbase-cdc/src/test/java/org/apache/flink/cdc/connectors/oceanbase/OceanBaseTestBase.java

@@ -20,7 +20,7 @@
 import org.apache.flink.cdc.connectors.oceanbase.testutils.OceanBaseCdcMetadata;
 import org.apache.flink.table.planner.factories.TestValuesTableFactory;
 import org.apache.flink.table.utils.LegacyRowResource;
-import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.test.util.AbstractTestBaseJUnit4;
 
 import org.junit.ClassRule;
 
@@ -43,7 +43,7 @@
 import static org.junit.Assert.assertTrue;
 
 /** Basic class for testing OceanBase source. */
-public abstract class OceanBaseTestBase extends AbstractTestBase {
+public abstract class OceanBaseTestBase extends AbstractTestBaseJUnit4 {
 
     private static final Pattern COMMENT_PATTERN = Pattern.compile("^(.*)--.*$");
 

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/PostgresTestBase.java

@@ -21,7 +21,7 @@
 import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfigFactory;
 import org.apache.flink.cdc.connectors.postgres.testutils.UniqueDatabase;
 import org.apache.flink.table.planner.factories.TestValuesTableFactory;
-import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.test.util.AbstractTestBaseJUnit4;
 import org.apache.flink.types.Row;
 
 import io.debezium.config.Configuration;
@@ -63,7 +63,7 @@
  * Basic class for testing PostgreSQL source, this contains a PostgreSQL container which enables wal
  * log.
  */
-public abstract class PostgresTestBase extends AbstractTestBase {
+public abstract class PostgresTestBase extends AbstractTestBaseJUnit4 {
     private static final Logger LOG = LoggerFactory.getLogger(PostgresTestBase.class);
     public static final Pattern COMMENT_PATTERN = Pattern.compile("^(.*)--.*$");
     public static final String DEFAULT_DB = "postgres";

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/SqlServerTestBase.java

@@ -18,7 +18,7 @@
 package org.apache.flink.cdc.connectors.sqlserver;
 
 import org.apache.flink.table.planner.factories.TestValuesTableFactory;
-import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.test.util.AbstractTestBaseJUnit4;
 
 import org.awaitility.Awaitility;
 import org.awaitility.core.ConditionTimeoutException;
@@ -49,7 +49,7 @@
 import static org.junit.Assert.assertNotNull;
 
 /** Utility class for sqlserver tests. */
-public class SqlServerTestBase extends AbstractTestBase {
+public class SqlServerTestBase extends AbstractTestBaseJUnit4 {
     private static final Logger LOG = LoggerFactory.getLogger(SqlServerTestBase.class);
     private static final Pattern COMMENT_PATTERN = Pattern.compile("^(.*)--.*$");
 

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/test/java/org/apache/flink/cdc/connectors/tidb/TiDBTestBase.java

@@ -17,7 +17,7 @@
 
 package org.apache.flink.cdc.connectors.tidb;
 
-import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.test.util.AbstractTestBaseJUnit4;
 
 import com.alibaba.dcm.DnsCacheManipulator;
 import org.apache.commons.lang3.RandomUtils;
@@ -53,7 +53,7 @@
 import static org.junit.Assert.assertNotNull;
 
 /** Utility class for tidb tests. */
-public class TiDBTestBase extends AbstractTestBase {
+public class TiDBTestBase extends AbstractTestBaseJUnit4 {
     private static final Logger LOG = LoggerFactory.getLogger(TiDBTestBase.class);
     private static final Pattern COMMENT_PATTERN = Pattern.compile("^(.*)--.*$");
 

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-vitess-cdc/src/test/java/org/apache/flink/cdc/connectors/vitess/VitessTestBase.java

@@ -18,7 +18,7 @@
 package org.apache.flink.cdc.connectors.vitess;
 
 import org.apache.flink.cdc.connectors.vitess.container.VitessContainer;
-import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.test.util.AbstractTestBaseJUnit4;
 
 import org.junit.BeforeClass;
 import org.slf4j.Logger;
@@ -43,7 +43,7 @@
 import static org.junit.Assert.assertNotNull;
 
 /** Basic class for testing Vitess source, this contains a Vitess container. */
-public abstract class VitessTestBase extends AbstractTestBase {
+public abstract class VitessTestBase extends AbstractTestBaseJUnit4 {
 
     private static final Logger LOG = LoggerFactory.getLogger(VitessTestBase.class);
     private static final Pattern COMMENT_PATTERN = Pattern.compile("^(.*)--.*$");

flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/SerializerTestBase.java

@@ -50,9 +50,9 @@
 import java.util.List;
 import java.util.concurrent.CyclicBarrier;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -486,7 +486,9 @@ public void testDuplicate() throws Exception {
     // --------------------------------------------------------------------------------------------
 
     private void deepEquals(String message, T should, T is) {
-        assertThat(message, is, CustomEqualityMatcher.deeplyEquals(should).withChecker(checker));
+        assertThat(is)
+                .as(message)
+                .matches(CustomEqualityMatcher.deeplyEquals(should).withChecker(checker));
     }
 
     // --------------------------------------------------------------------------------------------
@@ -580,11 +582,12 @@ public void run() {
                         T serdeTestItem = serializer.deserialize(dataInputDeserializer);
                         T copySerdeTestItem = serializer.copy(serdeTestItem);
                         dataOutputSerializer.clear();
-
-                        assertThat(
-                                "Serialization/Deserialization cycle resulted in an object that are not equal to the original.",
-                                copySerdeTestItem,
-                                CustomEqualityMatcher.deeplyEquals(testItem).withChecker(checker));
+                        assertThat(copySerdeTestItem)
+                                .as(
+                                        "Serialization/Deserialization cycle resulted in an object that are not equal to the original.")
+                                .matches(
+                                        CustomEqualityMatcher.deeplyEquals(testItem)
+                                                .withChecker(checker));
 
                         // try to enforce some upper bound to the test time
                         if (System.nanoTime() >= endTimeNanos) {

pom.xml

@@ -67,8 +67,8 @@ limitations under the License.
         <flink.reuseForks>true</flink.reuseForks>
 
         <!-- dependencies versions -->
-        <flink.version>1.19.1</flink.version>
-        <flink.major.version>1.19</flink.major.version>
+        <flink.version>1.20.0</flink.version>
+        <flink.major.version>1.20</flink.major.version>
         <flink.shaded.version>17.0</flink.shaded.version>
         <debezium.version>1.9.8.Final</debezium.version>
         <tikv.version>3.2.0</tikv.version>