Update test and pom to support jdk17.

lvyanquan · lvyanquan · commit 5f3b0b7d604f · 2025-12-23T20:34:56.000+08:00
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchema.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchema.java
@@ -146,7 +146,10 @@ public byte[] serialize(Event event) {
         reuseGenericRowData.setField(
                 5,
                 new GenericArrayData(
-                        jsonSerializers.get(dataChangeEvent.tableId()).getSchema().primaryKeys()
+                        jsonSerializers
+                                .get(dataChangeEvent.tableId())
+                                .getSchema()
+                                .primaryKeys()
                                 .stream()
                                 .map(StringData::fromString)
                                 .toArray()));
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/pom.xml b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/pom.xml
@@ -32,7 +32,7 @@ limitations under the License.
         <paimon.version>1.2.0</paimon.version>
         <hadoop.version>2.8.5</hadoop.version>
         <hive.version>2.3.9</hive.version>
-        <mockito.version>3.4.6</mockito.version>
+        <mockito.version>3.12.4</mockito.version>
     </properties>
 
     <dependencies>
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonWriter.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonWriter.java
@@ -109,16 +109,17 @@ public PaimonWriter(
     public Collection<MultiTableCommittable> prepareCommit() {
         long startTime = System.currentTimeMillis();
         List<MultiTableCommittable> committables =
-                writes.entrySet()
-                        .parallelStream()
+                writes.entrySet().parallelStream()
                         .flatMap(
                                 entry -> {
                                     try {
                                         // here we set it to lastCheckpointId+1 to
                                         // avoid prepareCommit the same checkpointId with the first
                                         // round.
-                                        return entry.getValue()
-                                                .prepareCommit(false, lastCheckpointId + 1).stream()
+                                        return entry
+                                                .getValue()
+                                                .prepareCommit(false, lastCheckpointId + 1)
+                                                .stream()
                                                 .map(
                                                         committable ->
                                                                 MultiTableCommittable
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/io/debezium/connector/db2/Db2Connection.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/io/debezium/connector/db2/Db2Connection.java
@@ -399,7 +399,8 @@ public Table getTableSchemaFromChangeTable(Db2ChangeTable changeTable) throws SQ
         // final List<Column> columns = columnEditors.subList(CHANGE_TABLE_DATA_COLUMN_OFFSET,
         // columnEditors.size() - 1).stream()
         final List<Column> columns =
-                columnEditors.subList(CHANGE_TABLE_DATA_COLUMN_OFFSET, columnEditors.size())
+                columnEditors
+                        .subList(CHANGE_TABLE_DATA_COLUMN_OFFSET, columnEditors.size())
                         .stream()
                         .map(
                                 c ->
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java
@@ -137,9 +137,10 @@ private int getParallelism() {
                     "parallelism.default: 4",
                     "execution.checkpointing.interval: 300",
                     "state.backend.type: hashmap",
-                    "env.java.opts.all: -Doracle.jdbc.timezoneAsRegion=false",
+                    "env.java.default-opts.all: --add-exports=java.base/sun.net.util=ALL-UNNAMED --add-exports=java.rmi/sun.rmi.registry=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED --add-exports=java.security.jgss/sun.security.krb5=ALL-UNNAMED --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.text=ALL-UNNAMED --add-opens=java.base/java.time=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.locks=ALL-UNNAMED --add-opens=java.base/jdk.internal.loader=ALL-UNNAMED --add-opens=java.base/java.security=ALL-UNNAMED --add-exports=java.base/sun.net.www=ALL-UNNAMED -Doracle.jdbc.timezoneAsRegion=false",
                     "execution.checkpointing.savepoint-dir: file:///opt/flink",
                     "restart-strategy.type: off",
+                    "pekko.ask.timeout: 60s",
                     // Set off-heap memory explicitly to avoid "java.lang.OutOfMemoryError: Direct
                     // buffer memory" error.
                     "taskmanager.memory.task.off-heap.size: 128mb",
diff --git a/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/pom.xml b/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/pom.xml
@@ -212,7 +212,7 @@ limitations under the License.
                 <executions>
                     <execution>
                         <id>copy-jars</id>
-                        <phase>test</phase>
+                        <phase>package</phase>
                         <goals>
                             <goal>copy</goal>
                         </goals>
diff --git a/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/java/org/apache/flink/cdc/connectors/tests/utils/FlinkContainerTestEnvironment.java b/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/java/org/apache/flink/cdc/connectors/tests/utils/FlinkContainerTestEnvironment.java
@@ -308,7 +308,10 @@ private String copyAndGetContainerPath(GenericContainer<?> container, String fil
     }
 
     private String getFlinkDockerImageTag() {
-        return String.format("flink:%s-scala_2.12", flinkVersion);
+        if (System.getProperty("java.specification.version").equals("17")) {
+            return String.format("flink:%s-scala_2.12-java17", flinkVersion);
+        }
+        return String.format("flink:%s-scala_2.12-java11", flinkVersion);
     }
 
     protected String getJdbcConnectorResourceName() {
diff --git a/pom.xml b/pom.xml
@@ -78,7 +78,7 @@ limitations under the License.
         <version.awaitility>4.2.0</version.awaitility>
         <slf4j.version>1.7.36</slf4j.version>
         <log4j.version>2.17.1</log4j.version>
-        <spotless.version>2.27.1</spotless.version>
+        <spotless.version>2.43.0</spotless.version>
         <commons-lang3.version>3.12.0</commons-lang3.version>
         <json-path.version>2.7.0</json-path.version>
         <jackson.version>2.13.2</jackson.version>
@@ -473,7 +473,7 @@ limitations under the License.
                 <configuration>
                     <java>
                         <googleJavaFormat>
-                            <version>1.7</version>
+                            <version>1.10.0</version>
                             <style>AOSP</style>
                         </googleJavaFormat>
 
@@ -569,6 +569,15 @@ limitations under the License.
                          does not understand some time zone which used in Ubuntu OS -->
                     <argLine>-Xms256m -Xmx2048m -Dmvn.forkNumber=${surefire.forkNumber}
                         -XX:+UseG1GC -Doracle.jdbc.timezoneAsRegion=false
+                        --add-opens=java.base/java.lang=ALL-UNNAMED
+                        --add-opens=java.base/java.lang.reflect=ALL-UNNAMED
+                        --add-opens=java.base/java.util.concurrent=ALL-UNNAMED
+                        --add-opens=java.base/java.util=ALL-UNNAMED
+                        --add-opens=java.sql/java.sql=ALL-UNNAMED
+                        --add-opens=java.base/java.nio=ALL-UNNAMED
+                        --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED
+                        --add-opens java.base/java.net=ALL-UNNAMED
+                        --add-exports=java.base/sun.nio.ch=ALL-UNNAMED
                     </argLine>
                 </configuration>
             </plugin>

Original file line number	Diff line number	Diff line change
`@@ -308,7 +308,10 @@ private String copyAndGetContainerPath(GenericContainer<?> container, String fil`
`308`	`308`	`}`
`309`	`309`
`310`	`310`	`private String getFlinkDockerImageTag() {`
`311`		`- return String.format("flink:%s-scala_2.12", flinkVersion);`
	`311`	`+ if (System.getProperty("java.specification.version").equals("17")) {`
	`312`	`+ return String.format("flink:%s-scala_2.12-java17", flinkVersion);`
	`313`	`+ }`
	`314`	`+ return String.format("flink:%s-scala_2.12-java11", flinkVersion);`
`312`	`315`	`}`
`313`	`316`
`314`	`317`	`protected String getJdbcConnectorResourceName() {`