tmp code to verify 2.0-preview

yunfengzhou-hub · Nov 22, 2024 · 48db90f · 48db90f
1 parent fa33b76
commit 48db90f
Show file tree

Hide file tree

Showing 16 changed files with 72 additions and 41 deletions.
diff --git a/paimon-flink/paimon-flink-cdc/pom.xml b/paimon-flink/paimon-flink-cdc/pom.xml
@@ -34,7 +34,7 @@ under the License.
     <name>Paimon : Flink : CDC</name>
 
     <properties>
-        <flink.version>1.20.0</flink.version>
+        <flink.version>2.0-preview1</flink.version>
         <flink.cdc.version>3.1.1</flink.cdc.version>
         <flink.mongodb.cdc.version>3.1.1</flink.mongodb.cdc.version>
         <avro.version>1.11.4</avro.version>

diff --git a/paimon-flink/paimon-flink-common/pom.xml b/paimon-flink/paimon-flink-common/pom.xml
@@ -34,7 +34,7 @@ under the License.
     <name>Paimon : Flink : Common</name>
 
     <properties>
-        <flink.version>1.20.0</flink.version>
+        <flink.version>2.0-preview1</flink.version>
     </properties>
 
     <dependencies>
@@ -169,6 +169,12 @@ under the License.
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-test-utils</artifactId>
+            <version>2.0-preview1</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

diff --git a/...flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/ActionFactory.java b/...flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/ActionFactory.java
@@ -25,7 +25,7 @@
 import org.apache.paimon.utils.Preconditions;
 
 import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.api.java.utils.MultipleParameterTool;
+import org.apache.flink.util.MultipleParameterTool;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

diff --git a/...ink-common/src/main/java/org/apache/paimon/flink/action/MultipleParameterToolAdapter.java b/...ink-common/src/main/java/org/apache/paimon/flink/action/MultipleParameterToolAdapter.java
@@ -18,7 +18,7 @@
 
 package org.apache.paimon.flink.action;
 
-import org.apache.flink.api.java.utils.MultipleParameterTool;
+import org.apache.flink.util.MultipleParameterTool;
 
 import java.util.Collection;
 

diff --git a/...imon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryAddressRegister.java b/...imon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryAddressRegister.java
@@ -23,6 +23,7 @@
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.Table;
 
+import org.apache.flink.api.connector.sink2.InitContext;
 import org.apache.flink.api.connector.sink2.Sink;
 import org.apache.flink.api.connector.sink2.SinkWriter;
 import org.apache.flink.api.connector.sink2.WriterInitContext;
@@ -40,7 +41,9 @@ public QueryAddressRegister(Table table) {
         this.serviceManager = ((FileStoreTable) table).store().newServiceManager();
     }
 
-    @Override
+    /**
+     * Do not annotate with <code>@override</code> here to maintain compatibility with Flink 1.20-.
+     */
     public SinkWriter<InternalRow> createWriter(InitContext context) {
         return new QueryAddressRegisterSinkWriter(serviceManager);
     }

diff --git a/...flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LogSinkFunction.java b/...flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LogSinkFunction.java
@@ -20,7 +20,7 @@
 
 import org.apache.paimon.table.sink.SinkRecord;
 
-import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.sink.legacy.SinkFunction;
 
 /** Log {@link SinkFunction} with {@link WriteCallback}. */
 public interface LogSinkFunction extends SinkFunction<SinkRecord> {

diff --git a/...on-flink-common/src/main/java/org/apache/paimon/flink/sink/RowDataStoreWriteOperator.java b/...on-flink-common/src/main/java/org/apache/paimon/flink/sink/RowDataStoreWriteOperator.java
@@ -29,7 +29,7 @@
 import org.apache.flink.api.common.state.CheckpointListener;
 import org.apache.flink.runtime.state.StateInitializationContext;
 import org.apache.flink.runtime.state.StateSnapshotContext;
-import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.sink.legacy.SinkFunction;
 import org.apache.flink.streaming.api.graph.StreamConfig;
 import org.apache.flink.streaming.api.operators.InternalTimerService;
 import org.apache.flink.streaming.api.operators.Output;

diff --git a/...-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RescaleBucketITCase.java b/...-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RescaleBucketITCase.java
@@ -24,9 +24,9 @@
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.utils.SnapshotManager;
 
+import org.apache.flink.configuration.StateRecoveryOptions;
 import org.apache.flink.core.execution.JobClient;
 import org.apache.flink.core.execution.SavepointFormatType;
-import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
 import org.apache.flink.types.Row;
 import org.junit.jupiter.api.Test;
 
@@ -106,9 +106,7 @@ public void testSuspendAndRecoverAfterRescaleOverwrite() throws Exception {
         assertThat(batchSql("SELECT * FROM T3")).containsExactlyInAnyOrderElementsOf(committedData);
 
         // step5: resume streaming job
-        sEnv.getConfig()
-                .getConfiguration()
-                .set(SavepointConfigOptions.SAVEPOINT_PATH, savepointPath);
+        sEnv.getConfig().getConfiguration().set(StateRecoveryOptions.SAVEPOINT_PATH, savepointPath);
         JobClient resumedJobClient =
                 startJobAndCommitSnapshot(streamSql, snapshotAfterRescale.id());
         // stop job

diff --git a/...n-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SchemaChangeITCase.java b/...n-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SchemaChangeITCase.java
@@ -1008,7 +1008,7 @@ public void testAlterTableNonPhysicalColumn() {
                                 + "  `ts` TIMESTAMP(3),\n"
                                 + "  `ee` VARCHAR(2147483647) METADATA,\n"
                                 + "  WATERMARK FOR `ts` AS `ts`\n"
-                                + ") ")
+                                + ")")
                 .doesNotContain("schema");
     }
 

diff --git a/...imon-flink-common/src/test/java/org/apache/paimon/flink/source/IteratorSourcesITCase.java b/...imon-flink-common/src/test/java/org/apache/paimon/flink/source/IteratorSourcesITCase.java
@@ -18,10 +18,10 @@
 
 package org.apache.paimon.flink.source;
 
+import org.apache.commons.collections.IteratorUtils;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
 import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.datastream.DataStreamUtils;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.test.util.MiniClusterWithClientResource;
@@ -67,7 +67,7 @@ public void testParallelSourceExecution() throws Exception {
                         "iterator source");
 
         final List<RowData> result =
-                DataStreamUtils.collectBoundedStream(stream, "Iterator Source Test");
+                IteratorUtils.toList(stream.executeAndCollect("Iterator Source Test"));
 
         verifySequence(result, 1L, 1_000L);
     }

diff --git a/...k-common/src/test/java/org/apache/paimon/flink/source/operator/TestingSourceOperator.java b/...k-common/src/test/java/org/apache/paimon/flink/source/operator/TestingSourceOperator.java
@@ -39,6 +39,7 @@
 import org.apache.flink.runtime.state.StateInitializationContextImpl;
 import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
 import org.apache.flink.streaming.api.operators.SourceOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
 import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
 import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
 import org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask;
@@ -59,12 +60,14 @@ public class TestingSourceOperator<T> extends SourceOperator<T, SimpleSourceSpli
     private final int parallelism;
 
     public TestingSourceOperator(
+            StreamOperatorParameters<T> parameters,
             SourceReader<T, SimpleSourceSplit> reader,
             WatermarkStrategy<T> watermarkStrategy,
             ProcessingTimeService timeService,
             boolean emitProgressiveWatermarks) {
 
         this(
+                parameters,
                 reader,
                 watermarkStrategy,
                 timeService,
@@ -75,6 +78,7 @@ public TestingSourceOperator(
     }
 
     public TestingSourceOperator(
+            StreamOperatorParameters<T> parameters,
             SourceReader<T, SimpleSourceSplit> reader,
             WatermarkStrategy<T> watermarkStrategy,
             ProcessingTimeService timeService,
@@ -84,6 +88,7 @@ public TestingSourceOperator(
             boolean emitProgressiveWatermarks) {
 
         super(
+                parameters,
                 (context) -> reader,
                 eventGateway,
                 new SimpleSourceSplitSerializer(),
@@ -145,12 +150,18 @@ public static <T> SourceOperator<T, SimpleSourceSplit> createTestOperator(
 
         final SourceOperator<T, SimpleSourceSplit> sourceOperator =
                 new TestingSourceOperator<>(
-                        reader, watermarkStrategy, timeService, emitProgressiveWatermarks);
+                        new StreamOperatorParameters(
+                                new SourceOperatorStreamTask<Integer>(new DummyEnvironment()),
+                                new MockStreamConfig(new Configuration(), 1),
+                                new MockOutput<>(new ArrayList<>()),
+                                null,
+                                null,
+                                null),
+                        reader,
+                        watermarkStrategy,
+                        timeService,
+                        emitProgressiveWatermarks);
 
-        sourceOperator.setup(
-                new SourceOperatorStreamTask<Integer>(new DummyEnvironment()),
-                new MockStreamConfig(new Configuration(), 1),
-                new MockOutput<>(new ArrayList<>()));
         sourceOperator.initializeState(stateContext);
         sourceOperator.open();
 

diff --git a/...ink-common/src/test/java/org/apache/paimon/flink/util/MiniClusterWithClientExtension.java b/...ink-common/src/test/java/org/apache/paimon/flink/util/MiniClusterWithClientExtension.java
@@ -29,7 +29,6 @@
 import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
 import org.apache.flink.streaming.util.TestStreamEnvironment;
 import org.apache.flink.test.junit5.InjectClusterClient;
-import org.apache.flink.test.util.TestEnvironment;
 import org.junit.jupiter.api.extension.AfterAllCallback;
 import org.junit.jupiter.api.extension.AfterEachCallback;
 import org.junit.jupiter.api.extension.BeforeAllCallback;
@@ -167,17 +166,12 @@ private void registerEnv(InternalMiniClusterExtension internalMiniClusterExtensi
                         .getOptional(CoreOptions.DEFAULT_PARALLELISM)
                         .orElse(internalMiniClusterExtension.getNumberSlots());
 
-        TestEnvironment executionEnvironment =
-                new TestEnvironment(
-                        internalMiniClusterExtension.getMiniCluster(), defaultParallelism, false);
-        executionEnvironment.setAsContext();
         TestStreamEnvironment.setAsContext(
                 internalMiniClusterExtension.getMiniCluster(), defaultParallelism);
     }
 
     private void unregisterEnv(InternalMiniClusterExtension internalMiniClusterExtension) {
         TestStreamEnvironment.unsetAsContext();
-        TestEnvironment.unsetAsContext();
     }
 
     private MiniClusterClient createMiniClusterClient(

diff --git a/...aimon-flink-common/src/test/java/org/apache/paimon/flink/util/ReadWriteTableTestUtil.java b/...aimon-flink-common/src/test/java/org/apache/paimon/flink/util/ReadWriteTableTestUtil.java
@@ -23,8 +23,8 @@
 import org.apache.paimon.utils.BlockingIterator;
 
 import org.apache.flink.api.common.RuntimeExecutionMode;
-import org.apache.flink.api.common.restartstrategy.RestartStrategies;
-import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestartStrategyOptions;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.api.EnvironmentSettings;
 import org.apache.flink.table.api.TableEnvironment;
@@ -36,6 +36,7 @@
 import javax.annotation.Nullable;
 
 import java.nio.file.Paths;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -52,7 +53,7 @@
 /** Test util for {@link ReadWriteTableITCase}. */
 public class ReadWriteTableTestUtil {
 
-    private static final Time TIME_OUT = Time.seconds(10);
+    private static final Duration TIME_OUT = Duration.ofSeconds(10);
 
     public static final int DEFAULT_PARALLELISM = 2;
 
@@ -75,12 +76,16 @@ public static void init(String warehouse) {
     }
 
     public static void init(String warehouse, int parallelism) {
-        StreamExecutionEnvironment sExeEnv = buildStreamEnv(parallelism);
-        sExeEnv.getConfig().setRestartStrategy(RestartStrategies.noRestart());
+        StreamExecutionEnvironment sExeEnv =
+                buildStreamEnv(
+                        parallelism,
+                        RestartStrategyOptions.RestartStrategyType.NO_RESTART_STRATEGY);
         sEnv = StreamTableEnvironment.create(sExeEnv);
 
-        bExeEnv = buildBatchEnv(parallelism);
-        bExeEnv.getConfig().setRestartStrategy(RestartStrategies.noRestart());
+        bExeEnv =
+                buildBatchEnv(
+                        parallelism,
+                        RestartStrategyOptions.RestartStrategyType.NO_RESTART_STRATEGY);
         bEnv = StreamTableEnvironment.create(bExeEnv, EnvironmentSettings.inBatchMode());
 
         ReadWriteTableTestUtil.warehouse = warehouse;
@@ -95,16 +100,26 @@ public static void init(String warehouse, int parallelism) {
         bEnv.useCatalog(catalog);
     }
 
-    public static StreamExecutionEnvironment buildStreamEnv(int parallelism) {
-        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+    public static StreamExecutionEnvironment buildStreamEnv(
+            int parallelism, RestartStrategyOptions.RestartStrategyType restartStrategyType) {
+        Configuration configuration = new Configuration();
+        configuration.set(
+                RestartStrategyOptions.RESTART_STRATEGY, restartStrategyType.getMainValue());
+        final StreamExecutionEnvironment env =
+                StreamExecutionEnvironment.getExecutionEnvironment(configuration);
         env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
         env.enableCheckpointing(100);
         env.setParallelism(parallelism);
         return env;
     }
 
-    public static StreamExecutionEnvironment buildBatchEnv(int parallelism) {
-        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+    public static StreamExecutionEnvironment buildBatchEnv(
+            int parallelism, RestartStrategyOptions.RestartStrategyType restartStrategyType) {
+        Configuration configuration = new Configuration();
+        configuration.set(
+                RestartStrategyOptions.RESTART_STRATEGY, restartStrategyType.getMainValue());
+        final StreamExecutionEnvironment env =
+                StreamExecutionEnvironment.getExecutionEnvironment(configuration);
         env.setRuntimeMode(RuntimeExecutionMode.BATCH);
         env.setParallelism(parallelism);
         return env;
@@ -270,7 +285,7 @@ public static void testBatchRead(String query, List<Row> expected) throws Except
         try (BlockingIterator<Row, Row> iterator = BlockingIterator.of(resultItr)) {
             if (!expected.isEmpty()) {
                 List<Row> result =
-                        iterator.collect(expected.size(), TIME_OUT.getSize(), TIME_OUT.getUnit());
+                        iterator.collect(expected.size(), TIME_OUT.getSeconds(), TimeUnit.SECONDS);
                 assertThat(toInsertOnlyRows(result))
                         .containsExactlyInAnyOrderElementsOf(toInsertOnlyRows(expected));
             }

diff --git a/paimon-flink/paimon-flink-common/src/test/resources/log4j2-test.properties b/paimon-flink/paimon-flink-common/src/test/resources/log4j2-test.properties
@@ -18,7 +18,7 @@
 
 # Set root logger level to OFF to not flood build logs
 # set manually to INFO for debugging purposes
-rootLogger.level = OFF
+rootLogger.level = INFO
 rootLogger.appenderRef.test.ref = TestLogger
 
 appender.testlogger.name = TestLogger

diff --git a/pom.xml b/pom.xml
@@ -95,7 +95,7 @@ under the License.
         <junit5.version>5.8.1</junit5.version>
         <spotless.version>2.13.0</spotless.version>
         <spotless.delimiter>package</spotless.delimiter>
-        <target.java.version>1.8</target.java.version>
+        <target.java.version>11</target.java.version>
         <assertj.version>3.23.1</assertj.version>
         <javax.annotation-api.version>1.3.2</javax.annotation-api.version>
         <flink.forkCount>1C</flink.forkCount>
@@ -836,6 +836,10 @@ under the License.
                             <!-- Prevents recompilation due to missing package-info.class, see MCOMPILER-205 -->
                             <arg>-Xpkginfo:always</arg>
                             <arg>-Xlint:deprecation</arg>
+                            <arg>--add-exports=java.base/sun.net.util=ALL-UNNAMED</arg>
+                            <arg>--add-exports=java.management/sun.management=ALL-UNNAMED</arg>
+                            <arg>--add-exports=java.rmi/sun.rmi.registry=ALL-UNNAMED</arg>
+                            <arg>--add-exports=java.security.jgss/sun.security.krb5=ALL-UNNAMED</arg>
                         </compilerArgs>
                     </configuration>
                 </plugin>

diff --git a/tools/ci/paimon-ci-tools/pom.xml b/tools/ci/paimon-ci-tools/pom.xml
@@ -82,8 +82,8 @@ under the License.
 				<groupId>org.apache.maven.plugins</groupId>
 				<artifactId>maven-compiler-plugin</artifactId>
 				<configuration>
-					<source>8</source>
-					<target>8</target>
+					<source>11</source>
+					<target>11</target>
 				</configuration>
 			</plugin>
 		</plugins>