[FLINK-39697] Bump Flink version to 2.2.1

.github/workflows/push_pr.yml

@@ -28,8 +28,8 @@ jobs:
   compile_and_test:
     strategy:
       matrix:
-        flink: [ 1.20.3 ]
-        jdk: [ '8, 11, 17' ]
+        flink: [ 2.2.1 ]
+        jdk: [ '11, 17' ]
     uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
     with:
       flink_version: ${{ matrix.flink }}

flink-connector-pulsar/pom.xml

@@ -133,14 +133,21 @@ under the License.
 			<artifactId>flink-table-test-utils</artifactId>
 			<version>${flink.version}</version>
 			<scope>test</scope>
+			<exclusions>
+				<exclusion>
+					<groupId>org.apache.flink</groupId>
+					<artifactId>flink-table-planner-loader</artifactId>
+				</exclusion>
+			</exclusions>
 		</dependency>
 
 		<dependency>
 			<groupId>org.apache.flink</groupId>
-			<artifactId>flink-table-api-scala-bridge_${scala.binary.version}</artifactId>
+			<artifactId>flink-table-planner_${scala.binary.version}</artifactId>
 			<version>${flink.version}</version>
 			<scope>test</scope>
 		</dependency>
+
 		<dependency>
 			<groupId>org.apache.flink</groupId>
 			<artifactId>flink-table-planner_${scala.binary.version}</artifactId>

flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeInformation.java

@@ -19,7 +19,7 @@
 package org.apache.flink.connector.pulsar.common.schema;
 
 import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.serialization.SerializerConfig;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 
@@ -67,7 +67,7 @@ public boolean isKeyType() {
     }
 
     @Override
-    public TypeSerializer<T> createSerializer(ExecutionConfig config) {
+    public TypeSerializer<T> createSerializer(SerializerConfig config) {
         return new PulsarSchemaTypeSerializer<>(schema);
     }
 

flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaTypeSerializer.java

@@ -202,7 +202,7 @@ public TypeSerializer<T> restoreSerializer() {
 
         @Override
         public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
-                TypeSerializer<T> newSerializer) {
+                TypeSerializerSnapshot<T> oldSerializerSnapshot) {
             return TypeSerializerSchemaCompatibility.compatibleAsIs();
         }
     }

flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSink.java

@@ -21,7 +21,11 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.connector.sink2.Committer;
-import org.apache.flink.api.connector.sink2.TwoPhaseCommittingSink;
+import org.apache.flink.api.connector.sink2.CommitterInitContext;
+import org.apache.flink.api.connector.sink2.CommittingSinkWriter;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SupportsCommitter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
 import org.apache.flink.connector.base.DeliveryGuarantee;
 import org.apache.flink.connector.pulsar.common.crypto.PulsarCrypto;
 import org.apache.flink.connector.pulsar.sink.committer.PulsarCommittable;
@@ -81,7 +85,7 @@
  * @param <IN> The input type of the sink.
  */
 @PublicEvolving
-public class PulsarSink<IN> implements TwoPhaseCommittingSink<IN, PulsarCommittable> {
+public class PulsarSink<IN> implements Sink<IN>, SupportsCommitter<PulsarCommittable> {
     private static final long serialVersionUID = 4416714587951282119L;
 
     private final SinkConfiguration sinkConfiguration;
@@ -129,7 +133,7 @@ public static <IN> PulsarSinkBuilder<IN> builder() {
 
     @Internal
     @Override
-    public PrecommittingSinkWriter<IN, PulsarCommittable> createWriter(InitContext initContext)
+    public CommittingSinkWriter<IN, PulsarCommittable> createWriter(WriterInitContext initContext)
             throws PulsarClientException {
         return new PulsarWriter<>(
                 sinkConfiguration,
@@ -143,7 +147,7 @@ public PrecommittingSinkWriter<IN, PulsarCommittable> createWriter(InitContext i
 
     @Internal
     @Override
-    public Committer<PulsarCommittable> createCommitter() {
+    public Committer<PulsarCommittable> createCommitter(CommitterInitContext context) {
         return new PulsarCommitter(sinkConfiguration);
     }
 

flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/SinkConfiguration.java

@@ -19,7 +19,7 @@
 package org.apache.flink.connector.pulsar.sink.config;
 
 import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.connector.base.DeliveryGuarantee;
 import org.apache.flink.connector.pulsar.common.config.PulsarConfiguration;
@@ -80,9 +80,9 @@ public DeliveryGuarantee getDeliveryGuarantee() {
     /**
      * Pulsar's transactions have a timeout mechanism for the uncommitted transaction. We use
      * transactions for making sure the message could be written only once. Since the checkpoint
-     * interval couldn't be acquired from {@link InitContext}, we have to expose this option. Make
-     * sure this value is greater than the checkpoint interval. Create a pulsar producer builder by
-     * using the given Configuration.
+     * interval couldn't be acquired from {@link WriterInitContext}, we have to expose this option.
+     * Make sure this value is greater than the checkpoint interval. Create a pulsar producer
+     * builder by using the given Configuration.
      */
     public long getTransactionTimeoutMillis() {
         return transactionTimeoutMillis;

flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java

@@ -22,8 +22,9 @@
 import org.apache.flink.api.common.operators.MailboxExecutor;
 import org.apache.flink.api.common.operators.ProcessingTimeService;
 import org.apache.flink.api.common.serialization.SerializationSchema.InitializationContext;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
-import org.apache.flink.api.connector.sink2.TwoPhaseCommittingSink.PrecommittingSinkWriter;
+import org.apache.flink.api.connector.sink2.CommittingSinkWriter;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
 import org.apache.flink.connector.base.DeliveryGuarantee;
 import org.apache.flink.connector.pulsar.common.crypto.PulsarCrypto;
 import org.apache.flink.connector.pulsar.sink.committer.PulsarCommittable;
@@ -67,7 +68,7 @@
  * @param <IN> The type of the input elements.
  */
 @Internal
-public class PulsarWriter<IN> implements PrecommittingSinkWriter<IN, PulsarCommittable> {
+public class PulsarWriter<IN> implements CommittingSinkWriter<IN, PulsarCommittable> {
     private static final Logger LOG = LoggerFactory.getLogger(PulsarWriter.class);
 
     private final PulsarSerializationSchema<IN> serializationSchema;
@@ -101,7 +102,7 @@ public PulsarWriter(
             TopicRouter<IN> topicRouter,
             MessageDelayer<IN> messageDelayer,
             PulsarCrypto pulsarCrypto,
-            InitContext initContext)
+            WriterInitContext initContext)
             throws PulsarClientException {
         checkNotNull(sinkConfiguration);
         this.serializationSchema = checkNotNull(serializationSchema);
@@ -139,7 +140,8 @@ public PulsarWriter(
     }
 
     @Override
-    public void write(IN element, Context context) throws IOException, InterruptedException {
+    public void write(IN element, SinkWriter.Context context)
+            throws IOException, InterruptedException {
         PulsarMessage<?> message = serializationSchema.serialize(element, sinkContext);
 
         // Choose the right topic to send.
@@ -185,7 +187,8 @@ private void throwSendingException(String topic, Throwable ex) {
 
     @SuppressWarnings({"rawtypes", "unchecked"})
     private TypedMessageBuilder<?> createMessageBuilder(
-            String topic, Context context, PulsarMessage<?> message) throws PulsarClientException {
+            String topic, SinkWriter.Context context, PulsarMessage<?> message)
+            throws PulsarClientException {
 
         Schema<?> schema = message.getSchema();
         TypedMessageBuilder<?> builder = producerRegister.createMessageBuilder(topic, schema);

flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/context/PulsarSinkContextImpl.java

@@ -20,7 +20,7 @@
 
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.operators.ProcessingTimeService;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
 import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
 import org.apache.flink.connector.pulsar.sink.writer.topic.MetadataListener;
 import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicMetadata;
@@ -38,11 +38,11 @@ public class PulsarSinkContextImpl implements PulsarSinkContext {
     private final MetadataListener metadataListener;
 
     public PulsarSinkContextImpl(
-            InitContext initContext,
+            WriterInitContext initContext,
             SinkConfiguration sinkConfiguration,
             MetadataListener metadataListener) {
-        this.parallelInstanceId = initContext.getSubtaskId();
-        this.numberOfParallelSubtasks = initContext.getNumberOfParallelSubtasks();
+        this.parallelInstanceId = initContext.getTaskInfo().getIndexOfThisSubtask();
+        this.numberOfParallelSubtasks = initContext.getTaskInfo().getNumberOfParallelSubtasks();
         this.processingTimeService = initContext.getProcessingTimeService();
         this.enableSchemaEvolution = sinkConfiguration.isEnableSchemaEvolution();
         this.metadataListener = metadataListener;

flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java

@@ -19,8 +19,8 @@
 package org.apache.flink.connector.pulsar.source;
 
 import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializerConfig;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.connector.source.Boundedness;
 import org.apache.flink.configuration.ConfigOption;
@@ -434,7 +434,7 @@ public <K, V, T extends OUT> PulsarSourceBuilder<T> setDeserializationSchema(
      * only used for treating messages that was written into pulsar by {@link TypeInformation}.
      */
     public <T extends OUT> PulsarSourceBuilder<T> setDeserializationSchema(
-            TypeInformation<T> information, ExecutionConfig config) {
+            TypeInformation<T> information, SerializerConfig config) {
         return setDeserializationSchema(new PulsarTypeInformationWrapper<>(information, config));
     }
 

flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceFetcherManager.java

@@ -20,12 +20,9 @@
 
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
-import org.apache.flink.connector.base.source.reader.SourceReaderBase;
 import org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher;
 import org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager;
 import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
-import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
 import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
 import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
 
@@ -59,16 +56,13 @@ public class PulsarSourceFetcherManager
     /**
      * Creates a new SplitFetcherManager with multiple I/O threads.
      *
-     * @param elementsQueue The queue that is used to hand over data from the I/O thread (the
-     *     fetchers) to the reader, which emits the records and book-keeps the state. This must be
-     *     the same queue instance that is also passed to the {@link SourceReaderBase}.
      * @param splitReaderSupplier The factory for the split reader that connects to the source
+     * @param configuration The configuration for the fetcher manager
      */
     public PulsarSourceFetcherManager(
-            FutureCompletingBlockingQueue<RecordsWithSplitIds<Message<byte[]>>> elementsQueue,
             Supplier<SplitReader<Message<byte[]>, PulsarPartitionSplit>> splitReaderSupplier,
             Configuration configuration) {
-        super(elementsQueue, splitReaderSupplier, configuration);
+        super(splitReaderSupplier, configuration);
     }
 
     /**
@@ -86,7 +80,7 @@ public void addSplits(List<PulsarPartitionSplit> splitsToAdd) {
         }
     }
 
-    // @Override // to keep compatible with Flink 1.17
+    @Override
     public void removeSplits(List<PulsarPartitionSplit> splitsToRemove) {
         // TODO empty - wait for FLINK-31748 to implement it.
     }

flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReader.java

@@ -22,10 +22,8 @@
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.connector.source.ReaderOutput;
 import org.apache.flink.api.connector.source.SourceReaderContext;
-import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
 import org.apache.flink.connector.base.source.reader.SourceReaderBase;
 import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
-import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
 import org.apache.flink.connector.pulsar.common.crypto.PulsarCrypto;
 import org.apache.flink.connector.pulsar.common.schema.BytesSchema;
 import org.apache.flink.connector.pulsar.common.schema.PulsarSchema;
@@ -84,14 +82,12 @@ public class PulsarSourceReader<OUT>
     private ScheduledExecutorService cursorScheduler;
 
     private PulsarSourceReader(
-            FutureCompletingBlockingQueue<RecordsWithSplitIds<Message<byte[]>>> elementsQueue,
             PulsarSourceFetcherManager fetcherManager,
             PulsarDeserializationSchema<OUT> deserializationSchema,
             SourceConfiguration sourceConfiguration,
             PulsarClient pulsarClient,
             SourceReaderContext context) {
         super(
-                elementsQueue,
                 fetcherManager,
                 new PulsarRecordEmitter<>(deserializationSchema),
                 sourceConfiguration,
@@ -252,11 +248,6 @@ public static <OUT> PulsarSourceReader<OUT> create(
             SourceReaderContext readerContext)
             throws Exception {
 
-        // Create a message queue with the predefined source option.
-        int queueCapacity = sourceConfiguration.getMessageQueueCapacity();
-        FutureCompletingBlockingQueue<RecordsWithSplitIds<Message<byte[]>>> elementsQueue =
-                new FutureCompletingBlockingQueue<>(queueCapacity);
-
         PulsarClient pulsarClient = createClient(sourceConfiguration);
 
         // Initialize the deserialization schema before creating the pulsar reader.
@@ -287,10 +278,9 @@ public static <OUT> PulsarSourceReader<OUT> create(
 
         PulsarSourceFetcherManager fetcherManager =
                 new PulsarSourceFetcherManager(
-                        elementsQueue, splitReaderSupplier, readerContext.getConfiguration());
+                        splitReaderSupplier, readerContext.getConfiguration());
 
         return new PulsarSourceReader<>(
-                elementsQueue,
                 fetcherManager,
                 deserializationSchema,
                 sourceConfiguration,

flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/deserializer/GenericRecordDeserializationSchema.java

@@ -1,3 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package org.apache.flink.connector.pulsar.source.reader.deserializer;
 
 import org.apache.flink.annotation.Internal;

flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarDeserializationSchema.java

@@ -70,7 +70,7 @@ default void open(PulsarInitializationContext context, SourceConfiguration confi
 
     /** An interface for providing extra schema initial context for users. */
     @PublicEvolving
-    public interface PulsarInitializationContext extends InitializationContext {
+    interface PulsarInitializationContext extends InitializationContext {
 
         /** Return the internal client for extra dynamic features. */
         PulsarClient getPulsarClient();

flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarTypeInformationWrapper.java

@@ -18,19 +18,17 @@
 package org.apache.flink.connector.pulsar.source.reader.deserializer;
 
 import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.serialization.SerializerConfig;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.core.memory.DataInputDeserializer;
 import org.apache.flink.util.Collector;
 
 import org.apache.pulsar.client.api.Message;
 
 /**
  * Wrap the flink TypeInformation into a {@code PulsarDeserializationSchema}. We would create a
- * flink {@code TypeSerializer} by using given ExecutionConfig. This execution config could be
- * {@link ExecutionEnvironment#getConfig()}.
+ * flink {@code TypeSerializer} by using given SerializerConfig.
  */
 @Internal
 public class PulsarTypeInformationWrapper<T> implements PulsarDeserializationSchema<T> {
@@ -47,7 +45,7 @@ public class PulsarTypeInformationWrapper<T> implements PulsarDeserializationSch
     private final TypeInformation<T> information;
     private final TypeSerializer<T> serializer;
 
-    public PulsarTypeInformationWrapper(TypeInformation<T> information, ExecutionConfig config) {
+    public PulsarTypeInformationWrapper(TypeInformation<T> information, SerializerConfig config) {
         this.information = information;
         this.serializer = information.createSerializer(config);
     }