Update to Flink 2.0

Author: Grzegorz Kołakowski

.github/workflows/build.yml

@@ -18,14 +18,14 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        flink: ["1.18.1", "1.19.1", "1.20.0"]
+        flink: ["2.0.0"]
     steps:
       - uses: actions/checkout@v3
 
-      - name: Set up JDK 11
+      - name: Set up JDK 17
         uses: actions/setup-java@v3
         with:
-          java-version: '11'
+          java-version: '17'
           distribution: 'adopt'
           cache: maven
 
@@ -38,7 +38,7 @@ jobs:
 
       - name: Test JavaDoc
         run: mvn $MAVEN_CLI_OPTS $JAVA_ADDITIONAL_OPTS javadoc:javadoc
-        if: startsWith(matrix.flink, '1.20')
+        if: startsWith(matrix.flink, '2.0')
 
       - name: Add coverage to PR
         id: jacoco
@@ -48,4 +48,4 @@ jobs:
           token: ${{ secrets.GITHUB_TOKEN }}
           min-coverage-overall: 40
           min-coverage-changed-files: 60
-        if: startsWith(matrix.flink, '1.20') && github.event.pull_request.head.repo.fork == false
+        if: startsWith(matrix.flink, '2.0') && github.event.pull_request.head.repo.fork == false

.github/workflows/prepare_release_branch.yml

@@ -24,10 +24,10 @@ jobs:
     steps:
       - uses: actions/checkout@v3
 
-      - name: Set up JDK 11
+      - name: Set up JDK 17
         uses: actions/setup-java@v3
         with:
-          java-version: '11'
+          java-version: '17'
           distribution: 'adopt'
           cache: maven
 

.github/workflows/publish.yml

@@ -19,10 +19,10 @@ jobs:
       - name: Check release tag match  # ... and fail fast if they do not
         run: diff <(echo "${{ github.ref_name }}") <(echo "$(mvn -B help:evaluate -Dexpression=project.version -q -DforceStdout)")
 
-      - name: Set up JDK 11
+      - name: Set up JDK 17
         uses: actions/setup-java@v4
         with:
-          java-version: '11'
+          java-version: '17'
           distribution: 'temurin'
           cache: maven
 
@@ -36,7 +36,7 @@ jobs:
       - name: Set up Apache Maven Central
         uses: actions/setup-java@v4
         with:
-          java-version: '11'
+          java-version: '17'
           distribution: 'temurin'
           server-id: central
           server-username: MAVEN_USERNAME

pom.xml

@@ -58,15 +58,15 @@ under the License.
     <!-- IMPORTANT: If you update Flink, remember to update link to its docs in maven-javadoc-plugin <links>
          section, omitting the patch part (so for 1.15.0 use 1.15). -->
 
-    <flink.version>1.18.1</flink.version>
+    <flink.version>2.0.0</flink.version>
 
-    <target.java.version>11</target.java.version>
+    <target.java.version>17</target.java.version>
     <scala.binary.version>2.12</scala.binary.version>
     <maven.compiler.source>${target.java.version}</maven.compiler.source>
     <maven.compiler.target>${target.java.version}</maven.compiler.target>
     <lombok.version>1.18.22</lombok.version>
     <jackson.version>2.18.1</jackson.version>
-    <junit5.version>5.10.1</junit5.version>
+    <junit5.version>5.11.4</junit5.version>
     <junit.jupiter.version>${junit5.version}</junit.jupiter.version>
     <assertj.core.version>3.21.0</assertj.core.version>
     <mockito.version>4.0.0</mockito.version>
@@ -95,12 +95,6 @@ under the License.
   <dependencies>
     <!-- Apache Flink dependencies -->
     <!-- These dependencies are provided, because they should not be packaged into the JAR file. -->
-    <dependency>
-      <groupId>org.apache.flink</groupId>
-      <artifactId>flink-java</artifactId>
-      <version>${flink.version}</version>
-      <scope>provided</scope>
-    </dependency>
     <dependency>
       <groupId>org.apache.flink</groupId>
       <artifactId>flink-clients</artifactId>
@@ -502,7 +496,7 @@ under the License.
             <version>3.6.3</version>
             <configuration>
               <links>
-                <link>https://nightlies.apache.org/flink/flink-docs-release-1.15/api/java/</link>
+                <link>https://nightlies.apache.org/flink/flink-docs-release-2.0/api/java/</link>
               </links>
             </configuration>
             <executions>

src/main/java/com/getindata/connectors/http/SchemaLifecycleAwareElementConverter.java

@@ -1,12 +1,12 @@
 package com.getindata.connectors.http;
 
 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.WriterInitContext;
 import org.apache.flink.connector.base.sink.writer.ElementConverter;
 
 /**
- * An enhancement for Flink's {@link ElementConverter} that expose {@link #open(InitContext)} method
- * that will be called by HTTP connect code to ensure that element converter is initialized
+ * An enhancement for Flink's {@link ElementConverter} that expose {@link #open(WriterInitContext)}
+ * method that will be called by HTTP connect code to ensure that element converter is initialized
  * properly. This is required for cases when Flink's SerializationSchema and DeserializationSchema
  * objects like JsonRowDataSerializationSchema are used.
  * <p>
@@ -29,6 +29,6 @@ public interface SchemaLifecycleAwareElementConverter<InputT, RequestEntryT>
      *
      * @param context Contextual information that can be used during initialization.
      */
-    void open(InitContext context);
+    void open(WriterInitContext context);
 
 }

src/main/java/com/getindata/connectors/http/internal/sink/HttpSinkInternal.java

@@ -5,9 +5,12 @@
 import java.util.Collections;
 import java.util.Properties;
 
+import org.apache.flink.api.connector.sink2.StatefulSinkWriter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
 import org.apache.flink.connector.base.sink.AsyncSinkBase;
 import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
 import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration;
 import org.apache.flink.core.io.SimpleVersionedSerializer;
 import org.apache.flink.util.Preconditions;
 import org.apache.flink.util.StringUtils;
@@ -115,7 +118,7 @@ protected HttpSinkInternal(
 
     @Override
     public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> createWriter(
-            InitContext context) throws IOException {
+            WriterInitContext context) throws IOException {
 
         ElementConverter<InputT, HttpSinkRequestEntry> elementConverter = getElementConverter();
         if (elementConverter instanceof SchemaLifecycleAwareElementConverter) {
@@ -126,12 +129,7 @@ public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> cr
         return new HttpSinkWriter<>(
             elementConverter,
             context,
-            getMaxBatchSize(),
-            getMaxInFlightRequests(),
-            getMaxBufferedRequests(),
-            getMaxBatchSizeInBytes(),
-            getMaxTimeInBufferMS(),
-            getMaxRecordSizeInBytes(),
+            getAsyncSinkWriterConfiguration(),
             endpointUrl,
             sinkHttpClientBuilder.build(
                 properties,
@@ -146,19 +144,14 @@ public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> cr
 
     @Override
     public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> restoreWriter(
-                InitContext context,
+                WriterInitContext context,
                 Collection<BufferedRequestState<HttpSinkRequestEntry>> recoveredState)
             throws IOException {
 
         return new HttpSinkWriter<>(
             getElementConverter(),
             context,
-            getMaxBatchSize(),
-            getMaxInFlightRequests(),
-            getMaxBufferedRequests(),
-            getMaxBatchSizeInBytes(),
-            getMaxTimeInBufferMS(),
-            getMaxRecordSizeInBytes(),
+            getAsyncSinkWriterConfiguration(),
             endpointUrl,
             sinkHttpClientBuilder.build(
                 properties,
@@ -171,6 +164,17 @@ public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> re
         );
     }
 
+    private AsyncSinkWriterConfiguration getAsyncSinkWriterConfiguration() {
+        return AsyncSinkWriterConfiguration.builder()
+            .setMaxBatchSize(getMaxBatchSize())
+            .setMaxBatchSizeInBytes(getMaxBatchSizeInBytes())
+            .setMaxInFlightRequests(getMaxInFlightRequests())
+            .setMaxBufferedRequests(getMaxBufferedRequests())
+            .setMaxTimeInBufferMS(getMaxTimeInBufferMS())
+            .setMaxRecordSizeInBytes(getMaxRecordSizeInBytes())
+            .build();
+    }
+
     @Override
     public SimpleVersionedSerializer<BufferedRequestState<HttpSinkRequestEntry>>
             getWriterStateSerializer() {

src/main/java/com/getindata/connectors/http/internal/sink/HttpSinkWriter.java

@@ -1,18 +1,18 @@
 package com.getindata.connectors.http.internal.sink;
 
 import java.util.Collection;
-import java.util.Collections;
 import java.util.List;
 import java.util.Properties;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.function.Consumer;
 
 import lombok.extern.slf4j.Slf4j;
-import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
 import org.apache.flink.connector.base.sink.writer.AsyncSinkWriter;
 import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
 import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.base.sink.writer.ResultHandler;
+import org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration;
 import org.apache.flink.metrics.Counter;
 import org.apache.flink.util.concurrent.ExecutorThreadFactory;
 
@@ -47,20 +47,13 @@ public class HttpSinkWriter<InputT> extends AsyncSinkWriter<InputT, HttpSinkRequ
 
     public HttpSinkWriter(
             ElementConverter<InputT, HttpSinkRequestEntry> elementConverter,
-            Sink.InitContext context,
-            int maxBatchSize,
-            int maxInFlightRequests,
-            int maxBufferedRequests,
-            long maxBatchSizeInBytes,
-            long maxTimeInBufferMS,
-            long maxRecordSizeInBytes,
+            WriterInitContext context,
+            AsyncSinkWriterConfiguration writerConfiguration,
             String endpointUrl,
             SinkHttpClient sinkHttpClient,
             Collection<BufferedRequestState<HttpSinkRequestEntry>> bufferedRequestStates,
             Properties properties) {
-
-        super(elementConverter, context, maxBatchSize, maxInFlightRequests, maxBufferedRequests,
-            maxBatchSizeInBytes, maxTimeInBufferMS, maxRecordSizeInBytes, bufferedRequestStates);
+        super(elementConverter, context, writerConfiguration, bufferedRequestStates);
         this.endpointUrl = endpointUrl;
         this.sinkHttpClient = sinkHttpClient;
 
@@ -83,7 +76,7 @@ public HttpSinkWriter(
     @Override
     protected void submitRequestEntries(
             List<HttpSinkRequestEntry> requestEntries,
-            Consumer<List<HttpSinkRequestEntry>> requestResult) {
+            ResultHandler<HttpSinkRequestEntry> resultHandler) {
         var future = sinkHttpClient.putRequests(requestEntries, endpointUrl);
         future.whenCompleteAsync((response, err) -> {
             if (err != null) {
@@ -114,7 +107,7 @@ protected void submitRequestEntries(
                 //requestResult.accept(Collections.emptyList());
                 //}
             }
-            requestResult.accept(Collections.emptyList());
+            resultHandler.complete();
         }, sinkWriterThreadPool);
     }
 

src/main/java/com/getindata/connectors/http/internal/table/SerializationSchemaElementConverter.java

@@ -1,8 +1,8 @@
 package com.getindata.connectors.http.internal.table;
 
 import org.apache.flink.api.common.serialization.SerializationSchema;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
 import org.apache.flink.api.connector.sink2.SinkWriter.Context;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.util.FlinkRuntimeException;
 
@@ -27,7 +27,7 @@ public SerializationSchemaElementConverter(
     }
 
     @Override
-    public void open(InitContext context) {
+    public void open(WriterInitContext context) {
         if (!schemaOpened) {
             try {
                 serializationSchema.open(context.asSerializationSchemaInitializationContext());