[FLINK-36648][Connectors/Kafka] Bump Flink version to 2.0.0

This closes #140

Author: lvyanquan

.github/workflows/push_pr.yml

@@ -28,16 +28,9 @@ jobs:
   compile_and_test:
     strategy:
       matrix:
-        flink: [ 1.20.0 ]
-        jdk: [ '8, 11, 17, 21' ]
+        flink: [ 2.0.0 ]
+        jdk: [ '17, 21' ]
     uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
     with:
       flink_version: ${{ matrix.flink }}
       jdk_version: ${{ matrix.jdk }}
-  python_test:
-    strategy:
-      matrix:
-        flink: [ 1.20.0 ]
-    uses: apache/flink-connector-shared-utils/.github/workflows/python_ci.yml@ci_utils
-    with:
-      flink_version: ${{ matrix.flink }}

.github/workflows/weekly.yml

@@ -30,7 +30,7 @@ jobs:
     strategy:
       matrix:
         flink_branches: [{
-          flink: 1.20-SNAPSHOT,
+          flink: 2.1-SNAPSHOT,
           branch: main
         }, {
           flink: 1.19.1,
@@ -46,5 +46,5 @@ jobs:
     with:
       flink_version: ${{ matrix.flink_branches.flink }}
       connector_branch: ${{ matrix.flink_branches.branch }}
-      jdk_version: ${{ matrix.flink_branches.jdk || '8, 11, 17, 21' }}
+      jdk_version: ${{ matrix.flink_branches.jdk || '17, 21' }}
       run_dependency_convergence: false

flink-connector-kafka-e2e-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SmokeKafkaITCase.java

@@ -19,12 +19,12 @@
 package org.apache.flink.tests.util.kafka;
 
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MemorySize;
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.connector.kafka.testutils.KafkaUtil;
 import org.apache.flink.connector.testframe.container.FlinkContainers;
 import org.apache.flink.connector.testframe.container.FlinkContainersSettings;
 import org.apache.flink.connector.testframe.container.TestcontainersSettings;
-import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
 import org.apache.flink.test.resources.ResourceTestUtils;
 import org.apache.flink.test.util.JobSubmission;
 import org.apache.flink.util.TestLoggerExtension;
@@ -60,6 +60,7 @@
 import java.util.UUID;
 import java.util.stream.Collectors;
 
+import static org.apache.flink.configuration.CheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH;
 import static org.apache.flink.connector.kafka.testutils.KafkaUtil.createKafkaContainer;
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -99,8 +100,14 @@ class SmokeKafkaITCase {
     private static Configuration getConfiguration() {
         // modify configuration to have enough slots
         final Configuration flinkConfig = new Configuration();
-        flinkConfig.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, 3);
-        flinkConfig.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
+        flinkConfig.set(TaskManagerOptions.NUM_TASK_SLOTS, 3);
+        flinkConfig.set(ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
+        flinkConfig.set(
+                org.apache.flink.configuration.JobManagerOptions.TOTAL_PROCESS_MEMORY,
+                MemorySize.ofMebiBytes(1024));
+        flinkConfig.set(
+                org.apache.flink.configuration.TaskManagerOptions.TOTAL_PROCESS_MEMORY,
+                MemorySize.ofMebiBytes(1024));
         // Workaround for FLINK-36454 ; default config is entirely overwritten
         flinkConfig.setString(
                 "env.java.opts.all",

flink-connector-kafka-e2e-tests/flink-streaming-kafka-test-base/src/main/java/org/apache/flink/streaming/kafka/test/base/KafkaExampleUtil.java

@@ -17,9 +17,12 @@
 
 package org.apache.flink.streaming.kafka.test.base;
 
-import org.apache.flink.api.common.restartstrategy.RestartStrategies;
-import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestartStrategyOptions;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.util.ParameterTool;
+
+import java.time.Duration;
 
 /** The util class for kafka example. */
 public class KafkaExampleUtil {
@@ -40,8 +43,14 @@ public static StreamExecutionEnvironment prepareExecutionEnv(ParameterTool param
                             + "--group.id <some id>");
         }
 
-        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-        env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));
+        Configuration configuration = new Configuration();
+        configuration.set(RestartStrategyOptions.RESTART_STRATEGY, "fixed-delay");
+        configuration.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, 4);
+        configuration.set(
+                RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_DELAY,
+                Duration.ofMillis(10000));
+        StreamExecutionEnvironment env =
+                StreamExecutionEnvironment.getExecutionEnvironment(configuration);
         env.enableCheckpointing(5000); // create a checkpoint every 5 seconds
         env.getConfig()
                 .setGlobalJobParameters(

flink-connector-kafka-e2e-tests/flink-streaming-kafka-test/src/main/java/org/apache/flink/streaming/kafka/test/KafkaExample.java

@@ -18,7 +18,6 @@
 package org.apache.flink.streaming.kafka.test;
 
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.api.java.utils.ParameterTool;
 import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
 import org.apache.flink.connector.kafka.sink.KafkaSink;
 import org.apache.flink.connector.kafka.source.KafkaSource;
@@ -27,6 +26,7 @@
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.kafka.test.base.KafkaExampleUtil;
+import org.apache.flink.util.ParameterTool;
 
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.IntegerDeserializer;

flink-connector-kafka/pom.xml

@@ -246,6 +246,12 @@ under the License.
             <version>${flink.version}</version>
             <scope>test</scope>
         </dependency>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-statebackend-forst</artifactId>
+			<version>${flink.version}</version>
+			<scope>test</scope>
+		</dependency>
 
         <!-- ArchUnit test dependencies -->
 

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/ExactlyOnceKafkaWriter.java

@@ -19,7 +19,7 @@
 
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.common.serialization.SerializationSchema;
-import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
 import org.apache.flink.connector.base.DeliveryGuarantee;
 import org.apache.flink.connector.kafka.sink.internal.BackchannelFactory;
 import org.apache.flink.connector.kafka.sink.internal.FlinkKafkaInternalProducer;
@@ -91,7 +91,7 @@ class ExactlyOnceKafkaWriter<IN> extends KafkaWriter<IN> {
             DeliveryGuarantee deliveryGuarantee,
             Properties kafkaProducerConfig,
             String transactionalIdPrefix,
-            Sink.InitContext sinkInitContext,
+            WriterInitContext sinkInitContext,
             KafkaRecordSerializationSchema<IN> recordSerializer,
             SerializationSchema.InitializationContext schemaContext,
             Collection<KafkaWriterState> recoveredStates) {

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSink.java

@@ -22,6 +22,7 @@
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.connector.sink2.Committer;
 import org.apache.flink.api.connector.sink2.CommitterInitContext;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
 import org.apache.flink.api.dag.Transformation;
 import org.apache.flink.connector.base.DeliveryGuarantee;
 import org.apache.flink.connector.kafka.lineage.KafkaDatasetFacet;
@@ -102,6 +103,7 @@ public static <IN> KafkaSinkBuilder<IN> builder() {
         return new KafkaSinkBuilder<>();
     }
 
+    @Internal
     @Override
     public Committer<KafkaCommittable> createCommitter(CommitterInitContext context) {
         return new KafkaCommitter(
@@ -119,14 +121,14 @@ public SimpleVersionedSerializer<KafkaCommittable> getCommittableSerializer() {
 
     @Internal
     @Override
-    public KafkaWriter<IN> createWriter(InitContext context) throws IOException {
+    public KafkaWriter<IN> createWriter(WriterInitContext context) throws IOException {
         return restoreWriter(context, Collections.emptyList());
     }
 
     @Internal
     @Override
     public KafkaWriter<IN> restoreWriter(
-            InitContext context, Collection<KafkaWriterState> recoveredState) {
+            WriterInitContext context, Collection<KafkaWriterState> recoveredState) {
         KafkaWriter<IN> writer;
         if (deliveryGuarantee == DeliveryGuarantee.EXACTLY_ONCE) {
             writer =

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java

@@ -21,7 +21,7 @@
 import org.apache.flink.api.common.operators.MailboxExecutor;
 import org.apache.flink.api.common.operators.ProcessingTimeService;
 import org.apache.flink.api.common.serialization.SerializationSchema;
-import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
 import org.apache.flink.connector.base.DeliveryGuarantee;
 import org.apache.flink.connector.kafka.MetricUtil;
 import org.apache.flink.connector.kafka.sink.internal.FlinkKafkaInternalProducer;
@@ -110,7 +110,7 @@ class KafkaWriter<IN>
     KafkaWriter(
             DeliveryGuarantee deliveryGuarantee,
             Properties kafkaProducerConfig,
-            Sink.InitContext sinkInitContext,
+            WriterInitContext sinkInitContext,
             KafkaRecordSerializationSchema<IN> recordSerializer,
             SerializationSchema.InitializationContext schemaContext) {
         this.deliveryGuarantee = checkNotNull(deliveryGuarantee, "deliveryGuarantee");
@@ -135,8 +135,8 @@ class KafkaWriter<IN>
         this.numRecordsOutErrorsCounter = metricGroup.getNumRecordsOutErrorsCounter();
         this.kafkaSinkContext =
                 new DefaultKafkaSinkContext(
-                        sinkInitContext.getSubtaskId(),
-                        sinkInitContext.getNumberOfParallelSubtasks(),
+                        sinkInitContext.getTaskInfo().getIndexOfThisSubtask(),
+                        sinkInitContext.getTaskInfo().getNumberOfParallelSubtasks(),
                         kafkaProducerConfig);
         try {
             recordSerializer.open(schemaContext, kafkaSinkContext);

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/TwoPhaseCommittingStatefulSink.java

@@ -1,14 +1,18 @@
 package org.apache.flink.connector.kafka.sink;
 
 import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.connector.sink2.StatefulSink;
-import org.apache.flink.api.connector.sink2.TwoPhaseCommittingSink;
+import org.apache.flink.api.connector.sink2.CommittingSinkWriter;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.StatefulSinkWriter;
+import org.apache.flink.api.connector.sink2.SupportsCommitter;
+import org.apache.flink.api.connector.sink2.SupportsWriterState;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
 
 import java.io.IOException;
 import java.util.Collection;
 
 /**
- * A combination of {@link TwoPhaseCommittingSink} and {@link StatefulSink}.
+ * A combination of {@link SupportsCommitter} and {@link SupportsWriterState}.
  *
  * <p>The purpose of this interface is to be able to pass an interface rather than a {@link
  * KafkaSink} implementation into the reducing sink which simplifies unit testing.
@@ -19,16 +23,15 @@
  */
 @Internal
 public interface TwoPhaseCommittingStatefulSink<InputT, WriterStateT, CommT>
-        extends TwoPhaseCommittingSink<InputT, CommT>, StatefulSink<InputT, WriterStateT> {
+        extends SupportsCommitter<CommT>, SupportsWriterState<InputT, WriterStateT>, Sink<InputT> {
 
-    PrecommittingStatefulSinkWriter<InputT, WriterStateT, CommT> createWriter(InitContext context)
-            throws IOException;
+    PrecommittingStatefulSinkWriter<InputT, WriterStateT, CommT> createWriter(
+            WriterInitContext context) throws IOException;
 
     PrecommittingStatefulSinkWriter<InputT, WriterStateT, CommT> restoreWriter(
-            InitContext context, Collection<WriterStateT> recoveredState) throws IOException;
+            WriterInitContext context, Collection<WriterStateT> recoveredState) throws IOException;
 
-    /** A combination of {@link PrecommittingSinkWriter} and {@link StatefulSinkWriter}. */
+    /** A combination of {@link StatefulSinkWriter}. */
     interface PrecommittingStatefulSinkWriter<InputT, WriterStateT, CommT>
-            extends PrecommittingSinkWriter<InputT, CommT>,
-                    StatefulSinkWriter<InputT, WriterStateT> {}
+            extends StatefulSinkWriter<InputT, WriterStateT>, CommittingSinkWriter<InputT, CommT> {}
 }