[FLINK-35490] Migrate tests to JUnit 5 and AssertJ

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

@@ -54,14 +54,14 @@ void takeoverOutput() {
     @Test
     void testNoArgument() throws Exception {
         CliFrontend.main(new String[] {});
-        assertThat(out.toString()).isEqualTo(HELP_MESSAGE);
+        assertThat(out).hasToString(HELP_MESSAGE);
         assertThat(err.toString()).isEmpty();
     }
 
     @Test
     void testGeneratingHelpMessage() throws Exception {
         CliFrontend.main(new String[] {"--help"});
-        assertThat(out.toString()).isEqualTo(HELP_MESSAGE);
+        assertThat(out).hasToString(HELP_MESSAGE);
         assertThat(err.toString()).isEmpty();
     }
 
@@ -84,9 +84,9 @@ void testGlobalPipelineConfigParsing() throws Exception {
                         flinkHome(),
                         "--global-config",
                         globalPipelineConfig());
-        assertThat(executor.getGlobalPipelineConfig().toMap().get("parallelism")).isEqualTo("1");
-        assertThat(executor.getGlobalPipelineConfig().toMap().get("schema.change.behavior"))
-                .isEqualTo("ignore");
+        assertThat(executor.getGlobalPipelineConfig().toMap())
+                .containsEntry("parallelism", "1")
+                .containsEntry("schema.change.behavior", "ignore");
     }
 
     @Test

flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/utils/ConfigurationUtilsTest.java

@@ -23,6 +23,7 @@
 
 import org.apache.flink.shaded.curator5.com.google.common.io.Resources;
 
+import org.assertj.core.api.Assertions;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
@@ -32,9 +33,6 @@
 import java.util.HashMap;
 import java.util.Map;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
 /** Unit test for {@link org.apache.flink.cdc.cli.utils.ConfigurationUtils}. */
 class ConfigurationUtilsTest {
 
@@ -81,8 +79,9 @@ void loadConfigFile(String resourcePath) throws Exception {
         for (Map.Entry<ConfigOption<?>, Object> entry : CONFIG_OPTIONS.entrySet()) {
             String key = entry.getKey().key();
             Object expectedValue = entry.getValue();
-            assertTrue(configMap.containsKey(key));
-            assertEquals(expectedValue, configMap.get(key));
+            Assertions.assertThat(configMap)
+                    .containsKey(key)
+                    .containsEntry(key, (String) expectedValue);
         }
     }
 }

flink-cdc-common/src/test/java/org/apache/flink/cdc/common/factories/FactoryHelperTests.java

@@ -33,7 +33,7 @@
 import java.util.Set;
 
 /** Tests for {@link FactoryHelper}. */
-public class FactoryHelperTests {
+class FactoryHelperTests {
 
     private Factory getDummyFactory() {
 

flink-cdc-common/src/test/java/org/apache/flink/cdc/common/schema/SelectorsTest.java

@@ -27,7 +27,7 @@
 class SelectorsTest {
 
     @Test
-    public void testTableSelector() {
+    void testTableSelector() {
 
         // nameSpace, schemaName, tableName
         Selectors selectors =

flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/ChangeEventUtilsTest.java

@@ -38,9 +38,9 @@
 import static org.apache.flink.cdc.common.testutils.assertions.EventAssertions.assertThat;
 
 /** A test for the {@link org.apache.flink.cdc.common.utils.ChangeEventUtils}. */
-public class ChangeEventUtilsTest {
+class ChangeEventUtilsTest {
     @Test
-    public void testResolveSchemaEvolutionOptions() {
+    void testResolveSchemaEvolutionOptions() {
 
         List<String> allTags =
                 Arrays.stream(SchemaChangeEventTypeFamily.ALL)
@@ -93,7 +93,7 @@ public void testResolveSchemaEvolutionOptions() {
     }
 
     @Test
-    public void testResolveSchemaEvolutionTag() {
+    void testResolveSchemaEvolutionTag() {
         assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("all"))
                 .isEqualTo(
                         Arrays.asList(

flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/SchemaMergingUtilsTest.java

@@ -1119,10 +1119,8 @@ void testGetLeastCommonType() {
     private static void assertTypeMergingVector(DataType incomingType, List<DataType> resultTypes) {
         Assertions.assertThat(ALL_TYPES)
                 .map(type -> getLeastCommonType(type, incomingType))
-                .containsExactlyElementsOf(resultTypes);
-
-        // Flip LHS and RHS should emit same outputs
-        Assertions.assertThat(ALL_TYPES)
+                .containsExactlyElementsOf(resultTypes)
+                // Flip LHS and RHS should emit same outputs
                 .map(type -> getLeastCommonType(incomingType, type))
                 .containsExactlyElementsOf(resultTypes);
     }

flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/SchemaUtilsTest.java

@@ -37,10 +37,10 @@
 import java.util.Map;
 
 /** A test for the {@link org.apache.flink.cdc.common.utils.SchemaUtils}. */
-public class SchemaUtilsTest {
+class SchemaUtilsTest {
 
     @Test
-    public void testApplyColumnSchemaChangeEvent() {
+    void testApplyColumnSchemaChangeEvent() {
         TableId tableId = TableId.parse("default.default.table1");
         Schema schema =
                 Schema.newBuilder()
@@ -165,7 +165,7 @@ public void testApplyColumnSchemaChangeEvent() {
     }
 
     @Test
-    public void testGetNumericPrecision() {
+    void testGetNumericPrecision() {
         Assertions.assertThat(SchemaUtils.getNumericPrecision(DataTypes.TINYINT())).isEqualTo(3);
         Assertions.assertThat(SchemaUtils.getNumericPrecision(DataTypes.SMALLINT())).isEqualTo(5);
         Assertions.assertThat(SchemaUtils.getNumericPrecision(DataTypes.INT())).isEqualTo(10);
@@ -180,7 +180,7 @@ public void testGetNumericPrecision() {
     }
 
     @Test
-    public void testInferWiderType() {
+    void testInferWiderType() {
         Assertions.assertThat(
                         SchemaUtils.inferWiderType(DataTypes.BINARY(17), DataTypes.BINARY(17)))
                 .isEqualTo(DataTypes.BINARY(17));
@@ -353,7 +353,7 @@ public void testInferWiderType() {
     }
 
     @Test
-    public void testInferWiderColumn() {
+    void testInferWiderColumn() {
         // Test normal merges
         Assertions.assertThat(
                         SchemaUtils.inferWiderColumn(
@@ -385,7 +385,7 @@ public void testInferWiderColumn() {
     }
 
     @Test
-    public void testInferWiderSchema() {
+    void testInferWiderSchema() {
         // Test normal merges
         Assertions.assertThat(
                         SchemaUtils.inferWiderSchema(

flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/StringUtilsTest.java

@@ -17,15 +17,15 @@
 
 package org.apache.flink.cdc.common.utils;
 
-import org.junit.Assert;
-import org.junit.Test;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
 
 /** A test for the {@link StringUtils}. */
-public class StringUtilsTest {
+class StringUtilsTest {
     @Test
-    public void testConvertToCamelCase() {
+    void testConvertToCamelCase() {
         String str = "AA_BB CC";
         String camelCaseStr = StringUtils.convertToCamelCase(str);
-        Assert.assertEquals("aaBbCc", camelCaseStr);
+        Assertions.assertThat(camelCaseStr).isEqualTo("aaBbCc");
     }
 }

flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkEnvironmentUtilsTest.java

@@ -23,31 +23,33 @@
 
 import org.apache.flink.shaded.guava31.com.google.common.collect.Lists;
 
-import org.junit.Assert;
-import org.junit.Test;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
 
 import java.net.URL;
 import java.util.Collections;
 import java.util.List;
 
 /** Test for {@link FlinkEnvironmentUtils}. */
-public class FlinkEnvironmentUtilsTest {
+class FlinkEnvironmentUtilsTest {
 
     @Test
-    public void testAddJars() throws Exception {
+    void testAddJars() throws Exception {
         Configuration configuration = new Configuration();
-        configuration.set(PipelineOptions.JARS, Collections.EMPTY_LIST);
+        configuration.set(PipelineOptions.JARS, Collections.emptyList());
         StreamExecutionEnvironment env =
                 StreamExecutionEnvironment.createLocalEnvironment(configuration);
 
         FlinkEnvironmentUtils.addJar(
                 env, Lists.newArrayList(new URL("file://a.jar"), new URL("file://a.jar")));
         List<String> expectedJars = Lists.newArrayList("file://a.jar");
-        Assert.assertEquals(expectedJars, env.getConfiguration().get(PipelineOptions.JARS));
+        Assertions.assertThat(env.getConfiguration().get(PipelineOptions.JARS))
+                .isEqualTo(expectedJars);
 
         FlinkEnvironmentUtils.addJar(
                 env, Lists.newArrayList(new URL("file://b.jar"), new URL("file://a.jar")));
         expectedJars.add("file://b.jar");
-        Assert.assertEquals(expectedJars, env.getConfiguration().get(PipelineOptions.JARS));
+        Assertions.assertThat(env.getConfiguration().get(PipelineOptions.JARS))
+                .isEqualTo(expectedJars);
     }
 }

flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerITCase.java

@@ -337,7 +337,7 @@ void testTransform(ValuesDataSink.SinkApi sinkApi) throws Exception {
                         sourceDef,
                         sinkDef,
                         Collections.emptyList(),
-                        new ArrayList<>(Arrays.asList(transformDef)),
+                        new ArrayList<>(Collections.singletonList(transformDef)),
                         Collections.emptyList(),
                         pipelineConfig);
 

flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerLenientITCase.java

@@ -382,7 +382,7 @@ void testTransform(ValuesDataSink.SinkApi sinkApi) throws Exception {
                         sourceDef,
                         sinkDef,
                         Collections.emptyList(),
-                        new ArrayList<>(Arrays.asList(transformDef)),
+                        new ArrayList<>(Collections.singletonList(transformDef)),
                         Collections.emptyList(),
                         pipelineConfig);
 

flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerTest.java

@@ -27,7 +27,7 @@
 
 import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
 
-import org.junit.jupiter.api.Assertions;
+import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 /** A test for the {@link FlinkPipelineComposer}. */
@@ -55,7 +55,8 @@ void testCreateDataSinkFromSinkDef() {
                                 new Configuration(),
                                 Thread.currentThread().getContextClassLoader()));
 
-        Assertions.assertTrue(dataSink instanceof DataSinkFactory1.TestDataSink);
-        Assertions.assertEquals("0.0.0.0", ((DataSinkFactory1.TestDataSink) dataSink).getHost());
+        Assertions.assertThat(dataSink).isExactlyInstanceOf(DataSinkFactory1.TestDataSink.class);
+        Assertions.assertThat(((DataSinkFactory1.TestDataSink) dataSink).getHost())
+                .isEqualTo("0.0.0.0");
     }
 }

flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineUdfITCase.java

@@ -59,7 +59,7 @@
 import static org.junit.jupiter.params.provider.Arguments.arguments;
 
 /** Integration test for UDFs. */
-public class FlinkPipelineUdfITCase {
+class FlinkPipelineUdfITCase {
     private static final int MAX_PARALLELISM = 4;
 
     // Always use parent-first classloader for CDC classes.

flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/translator/DataSinkTranslatorTest.java

@@ -29,7 +29,7 @@
 
 import org.apache.flink.shaded.guava31.com.google.common.collect.Lists;
 
-import org.junit.jupiter.api.Assertions;
+import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
@@ -60,7 +60,7 @@ void testPreWriteWithoutCommitSink() {
         OneInputTransformation<Event, Event> oneInputTransformation =
                 (OneInputTransformation) env.getTransformations().get(0);
         Transformation<?> reblanceTransformation = oneInputTransformation.getInputs().get(0);
-        Assertions.assertEquals(uid, reblanceTransformation.getUserProvidedNodeHash());
+        Assertions.assertThat(reblanceTransformation.getUserProvidedNodeHash()).isEqualTo(uid);
     }
 
     private static class EmptyEvent implements Event {}

flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/translator/DataSourceTranslatorTest.java

@@ -27,7 +27,7 @@
 
 import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
 
-import org.junit.jupiter.api.Assertions;
+import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 /** A test for the {@link DataSourceTranslator}. */
@@ -55,8 +55,9 @@ void testCreateDataSourceFromSourceDef() {
                                 new Configuration(),
                                 Thread.currentThread().getContextClassLoader()));
 
-        Assertions.assertTrue(dataSource instanceof DataSourceFactory1.TestDataSource);
-        Assertions.assertEquals(
-                "0.0.0.0", ((DataSourceFactory1.TestDataSource) dataSource).getHost());
+        Assertions.assertThat(dataSource)
+                .isExactlyInstanceOf(DataSourceFactory1.TestDataSource.class);
+        Assertions.assertThat(((DataSourceFactory1.TestDataSource) dataSource).getHost())
+                .isEqualTo("0.0.0.0");
     }
 }

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

@@ -41,9 +41,9 @@ limitations under the License.
 
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-runtime</artifactId>
-            <version>${flink.version}</version>
-            <scope>provided</scope>
+            <artifactId>flink-connector-test-util</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
         </dependency>
 
         <dependency>
@@ -64,6 +64,14 @@ limitations under the License.
                 </exclusion>
             </exclusions>
         </dependency>
+
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <version>${testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-test-utils-junit</artifactId>