[HTTP-37] - Add support for Flink 1.16. Modify CI/CD to run builds for Flink 1.15 and 1.16.

Signed-off-by: Krzysztof Chmielewski <krzysiek.chmielewski@gmail.com>

Author: kristoffSC

.github/workflows/build.yml

@@ -16,6 +16,9 @@ env:
 jobs:
   build:
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        flink: [ "1.15.0", "1.15.3", "1.16.1" ]
     steps:
       - uses: actions/checkout@v3
 
@@ -26,23 +29,24 @@ jobs:
           distribution: 'adopt'
           cache: maven
 
-      - name: Build
-        run: mvn $MAVEN_CLI_OPTS $JAVA_ADDITIONAL_OPTS compile
+      - name: Build for Flink ${{ matrix.flink }}
+        run: mvn $MAVEN_CLI_OPTS $JAVA_ADDITIONAL_OPTS -Dflink.version=${{ matrix.flink }} compile
 
-      - name: Tests
+      - name: Tests for Flink ${{ matrix.flink }}
         run: |
-          mvn $MAVEN_CLI_OPTS $JAVA_ADDITIONAL_OPTS test integration-test
+          mvn $MAVEN_CLI_OPTS $JAVA_ADDITIONAL_OPTS -Dflink.version=${{ matrix.flink }} test integration-test
           cat target/site/jacoco/index.html | grep -o 'Total[^%]*%'
 
       - name: Test JavaDoc
         run: mvn $MAVEN_CLI_OPTS $JAVA_ADDITIONAL_OPTS javadoc:javadoc
+        if: startsWith(matrix.flink, '1.15.0')
 
       - name: Add coverage to PR
         id: jacoco
         uses: madrapps/jacoco-report@v1.2
-        if: github.event_name == 'pull_request'
         with:
           paths: ${{ github.workspace }}/target/site/jacoco/jacoco.xml
           token: ${{ secrets.GITHUB_TOKEN }}
           min-coverage-overall: 40
           min-coverage-changed-files: 60
+        if: startsWith(matrix.flink, '1.15.0')

CHANGELOG.md

@@ -1,6 +1,9 @@
 # Changelog
 
 ## [Unreleased]
+-   Add support for Flink 1.16.
+-   Add [SchemaLifecycleAwareElementConverter](src/main/java/com/getindata/connectors/http/SchemaLifecycleAwareElementConverter.java) that can be used for createing
+    schema lifecycle aware Element converters for Http Sink.
 
 ## [0.8.1] - 2022-12-22
 

pom.xml

@@ -68,7 +68,7 @@ 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.15.0</flink.version>
+    <flink.version>[1.15.0,)</flink.version>
 
     <target.java.version>11</target.java.version>
     <scala.binary.version>2.12</scala.binary.version>

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

@@ -114,14 +114,29 @@ public HttpSinkBuilder<InputT> setSinkHttpClientBuilder(
     /**
      * @param elementConverter the {@link ElementConverter} to be used for the sink
      * @return {@link HttpSinkBuilder} itself
+     * @deprecated Converters set by this method might not work properly for Flink 1.16+. Use {@link
+     * #setElementConverter(SchemaLifecycleAwareElementConverter)} instead.
      */
+    @Deprecated
     @PublicEvolving
     public HttpSinkBuilder<InputT> setElementConverter(
         ElementConverter<InputT, HttpSinkRequestEntry> elementConverter) {
         this.elementConverter = elementConverter;
         return this;
     }
 
+    /**
+     * @param elementConverter the {@link SchemaLifecycleAwareElementConverter} to be used for the
+     *                         sink
+     * @return {@link HttpSinkBuilder} itself
+     */
+    @PublicEvolving
+    public HttpSinkBuilder<InputT> setElementConverter(
+        SchemaLifecycleAwareElementConverter<InputT, HttpSinkRequestEntry> elementConverter) {
+        this.elementConverter = elementConverter;
+        return this;
+    }
+
     public HttpSinkBuilder<InputT> setHttpPostRequestCallback(
         HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback) {
         this.httpPostRequestCallback = httpPostRequestCallback;

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

@@ -0,0 +1,34 @@
+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.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
+ * properly. This is required for cases when Flink's SerializationSchema and DeserializationSchema
+ * objects like JsonRowDataSerializationSchema are used.
+ * <p>
+ * This interface specifies the mapping between elements of a stream to request entries that can be
+ * sent to the destination. The mapping is provided by the end-user of a sink, not the sink
+ * creator.
+ *
+ * <p>The request entries contain all relevant information required to create and sent the actual
+ * request. Eg, for Kinesis Data Streams, the request entry includes the payload and the partition
+ * key.
+ */
+public interface SchemaLifecycleAwareElementConverter<InputT, RequestEntryT>
+    extends ElementConverter<InputT, RequestEntryT> {
+
+    /**
+     * Initialization element converter for the schema.
+     *
+     * <p>The provided {@link InitializationContext} can be used to access additional features such
+     * as e.g. registering user metrics.
+     *
+     * @param context Contextual information that can be used during initialization.
+     */
+    void open(InitContext context);
+
+}

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

@@ -13,6 +13,7 @@
 import org.apache.flink.util.StringUtils;
 
 import com.getindata.connectors.http.HttpPostRequestCallback;
+import com.getindata.connectors.http.SchemaLifecycleAwareElementConverter;
 import com.getindata.connectors.http.internal.HeaderPreprocessor;
 import com.getindata.connectors.http.internal.SinkHttpClientBuilder;
 
@@ -110,8 +111,14 @@ protected HttpSinkInternal(
     public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> createWriter(
             InitContext context) throws IOException {
 
+        ElementConverter<InputT, HttpSinkRequestEntry> elementConverter = getElementConverter();
+        if (elementConverter instanceof SchemaLifecycleAwareElementConverter) {
+            // This cast is needed for Flink 1.15.3 build
+            ((SchemaLifecycleAwareElementConverter<?, ?>) elementConverter).open(context);
+        }
+
         return new HttpSinkWriter<>(
-            getElementConverter(),
+            elementConverter,
             context,
             getMaxBatchSize(),
             getMaxInFlightRequests(),
@@ -128,8 +135,8 @@ public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> cr
 
     @Override
     public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> restoreWriter(
-            InitContext context,
-            Collection<BufferedRequestState<HttpSinkRequestEntry>> recoveredState)
+                InitContext context,
+                Collection<BufferedRequestState<HttpSinkRequestEntry>> recoveredState)
             throws IOException {
 
         return new HttpSinkWriter<>(

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

@@ -0,0 +1,47 @@
+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.table.data.RowData;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import com.getindata.connectors.http.SchemaLifecycleAwareElementConverter;
+import com.getindata.connectors.http.internal.sink.HttpSinkRequestEntry;
+
+public class SerializationSchemaElementConverter
+    implements SchemaLifecycleAwareElementConverter<RowData, HttpSinkRequestEntry> {
+
+    private final String insertMethod;
+
+    private final SerializationSchema<RowData> serializationSchema;
+
+    private boolean schemaOpened = false;
+
+    public SerializationSchemaElementConverter(
+        String insertMethod,
+        SerializationSchema<RowData> serializationSchema) {
+
+        this.insertMethod = insertMethod;
+        this.serializationSchema = serializationSchema;
+    }
+
+    @Override
+    public void open(InitContext context) {
+        if (!schemaOpened) {
+            try {
+                serializationSchema.open(context.asSerializationSchemaInitializationContext());
+                schemaOpened = true;
+            } catch (Exception e) {
+                throw new FlinkRuntimeException("Failed to initialize serialization schema.", e);
+            }
+        }
+    }
+
+    @Override
+    public HttpSinkRequestEntry apply(RowData rowData, Context context) {
+        return new HttpSinkRequestEntry(
+            insertMethod,
+            serializationSchema.serialize(rowData));
+    }
+}

src/main/java/com/getindata/connectors/http/internal/table/lookup/HttpTableLookupFunction.java

@@ -15,6 +15,7 @@
 
 import com.getindata.connectors.http.internal.PollingClient;
 import com.getindata.connectors.http.internal.PollingClientFactory;
+import com.getindata.connectors.http.internal.utils.SerializationSchemaUtils;
 
 @Slf4j
 public class HttpTableLookupFunction extends TableFunction<RowData> {
@@ -50,6 +51,11 @@ public HttpTableLookupFunction(
     @Override
     public void open(FunctionContext context) throws Exception {
         super.open(context);
+
+        this.responseSchemaDecoder.open(
+            SerializationSchemaUtils
+                .createDeserializationInitContext(HttpTableLookupFunction.class));
+
         this.localHttpCallCounter = new AtomicInteger(0);
         this.client = pollingClientFactory
             .createPollClient(options, responseSchemaDecoder);

src/main/java/com/getindata/connectors/http/internal/table/lookup/querycreators/GenericJsonQueryCreator.java

@@ -4,8 +4,10 @@
 
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.table.data.RowData;
+import org.apache.flink.util.FlinkRuntimeException;
 
 import com.getindata.connectors.http.LookupQueryCreator;
+import com.getindata.connectors.http.internal.utils.SerializationSchemaUtils;
 
 /**
  * A {@link LookupQueryCreator} that builds Json based body for REST requests, i.e. adds
@@ -17,6 +19,8 @@ public class GenericJsonQueryCreator implements LookupQueryCreator {
      */
     private final SerializationSchema<RowData> jsonSerialization;
 
+    private boolean schemaOpened = false;
+
     public GenericJsonQueryCreator(SerializationSchema<RowData> jsonSerialization) {
 
         this.jsonSerialization = jsonSerialization;
@@ -30,6 +34,22 @@ public GenericJsonQueryCreator(SerializationSchema<RowData> jsonSerialization) {
      */
     @Override
     public String createLookupQuery(RowData lookupDataRow) {
+        checkOpened();
         return new String(jsonSerialization.serialize(lookupDataRow), StandardCharsets.UTF_8);
     }
+
+    private void checkOpened() {
+        if (!schemaOpened) {
+            try {
+                jsonSerialization.open(
+                    SerializationSchemaUtils
+                        .createSerializationInitContext(GenericJsonQueryCreator.class));
+            } catch (Exception e) {
+                throw new FlinkRuntimeException(
+                    "Failed to initialize serialization schema for GenericJsonQueryCreatorFactory.",
+                    e);
+            }
+            schemaOpened = true;
+        }
+    }
 }

src/main/java/com/getindata/connectors/http/internal/table/sink/HttpDynamicSink.java

@@ -22,6 +22,7 @@
 import com.getindata.connectors.http.HttpSinkBuilder;
 import com.getindata.connectors.http.internal.sink.HttpSinkRequestEntry;
 import com.getindata.connectors.http.internal.sink.httpclient.JavaNetSinkHttpClient;
+import com.getindata.connectors.http.internal.table.SerializationSchemaElementConverter;
 import com.getindata.connectors.http.internal.utils.HttpHeaderUtils;
 import static com.getindata.connectors.http.internal.table.sink.HttpDynamicSinkConnectorOptions.INSERT_METHOD;
 import static com.getindata.connectors.http.internal.table.sink.HttpDynamicSinkConnectorOptions.URL;
@@ -128,10 +129,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
             .setHttpPostRequestCallback(httpPostRequestCallback)
             // In future header preprocessor could be set via custom factory
             .setHttpHeaderPreprocessor(HttpHeaderUtils.createDefaultHeaderPreprocessor())
-            .setElementConverter((rowData, _context) -> new HttpSinkRequestEntry(
-                insertMethod,
-                serializationSchema.serialize(rowData)
-            ))
+            .setElementConverter(
+                new SerializationSchemaElementConverter(insertMethod, serializationSchema))
             .setProperties(properties);
         addAsyncOptionsToSinkBuilder(builder);
 

src/main/java/com/getindata/connectors/http/internal/utils/SerializationSchemaUtils.java

@@ -0,0 +1,50 @@
+package com.getindata.connectors.http.internal.utils;
+
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.util.SimpleUserCodeClassLoader;
+import org.apache.flink.util.UserCodeClassLoader;
+
+public final class SerializationSchemaUtils {
+
+    private SerializationSchemaUtils() {
+
+    }
+
+    public static <T> org.apache.flink.api.common.serialization.SerializationSchema
+            .InitializationContext createSerializationInitContext(Class<T> classForClassLoader) {
+
+        return new org.apache.flink.api.common.serialization.SerializationSchema
+                .InitializationContext() {
+
+            @Override
+            public MetricGroup getMetricGroup() {
+                return new UnregisteredMetricsGroup();
+            }
+
+            @Override
+            public UserCodeClassLoader getUserCodeClassLoader() {
+                return SimpleUserCodeClassLoader.create(classForClassLoader.getClassLoader());
+            }
+        };
+    }
+
+    public static <T> org.apache.flink.api.common.serialization.DeserializationSchema
+        .InitializationContext createDeserializationInitContext(Class<T> classForClassLoader) {
+
+        return new org.apache.flink.api.common.serialization.DeserializationSchema
+            .InitializationContext() {
+
+            @Override
+            public MetricGroup getMetricGroup() {
+                return new UnregisteredMetricsGroup();
+            }
+
+            @Override
+            public UserCodeClassLoader getUserCodeClassLoader() {
+                return SimpleUserCodeClassLoader.create(classForClassLoader.getClassLoader());
+            }
+        };
+    }
+
+}

src/test/java/com/getindata/connectors/http/internal/sink/HttpSinkWriterTest.java

@@ -13,6 +13,7 @@
 import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
 import org.apache.flink.connector.base.sink.writer.ElementConverter;
 import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.groups.OperatorIOMetricGroup;
 import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -42,15 +43,20 @@ class HttpSinkWriterTest {
     @Mock
     private SinkHttpClient httpClient;
 
-    @Mock
+    // To work with Flink 1.15 and Flink 1.16
+    @Mock(lenient = true)
     private SinkWriterMetricGroup metricGroup;
 
+    @Mock
+    private OperatorIOMetricGroup operatorIOMetricGroup;
+
     @Mock
     private Counter errorCounter;
 
     @BeforeEach
     public void setUp() {
         when(metricGroup.getNumRecordsSendErrorsCounter()).thenReturn(errorCounter);
+        when(metricGroup.getIOMetricGroup()).thenReturn(operatorIOMetricGroup);
         when(context.metricGroup()).thenReturn(metricGroup);
 
         Collection<BufferedRequestState<HttpSinkRequestEntry>> stateBuffer = new ArrayList<>();

src/test/java/com/getindata/connectors/http/internal/table/lookup/JavaNetHttpPollingClientConnectionTest.java

@@ -49,6 +49,7 @@
 import com.getindata.connectors.http.internal.table.lookup.querycreators.GenericGetQueryCreator;
 import com.getindata.connectors.http.internal.table.lookup.querycreators.GenericJsonQueryCreator;
 import com.getindata.connectors.http.internal.utils.HttpHeaderUtils;
+import com.getindata.connectors.http.internal.utils.SerializationSchemaUtils;
 import static com.getindata.connectors.http.TestHelper.readTestFile;
 import static com.getindata.connectors.http.internal.table.lookup.HttpLookupTableSourceFactory.row;
 
@@ -368,6 +369,14 @@ private JavaNetHttpPollingClient setUpPollingClient(
                 .createDecodingFormat(dynamicTableFactoryContext, new Configuration())
                 .createRuntimeDecoder(dynamicTableSourceContext, physicalDataType);
 
+        try {
+            schemaDecoder.open(
+                SerializationSchemaUtils.createDeserializationInitContext(
+                    JavaNetHttpPollingClientConnectionTest.class));
+        } catch (Exception e) {
+            throw new RuntimeException("Unable to open schema decoder: " + e.getMessage(), e);
+        }
+
         JavaNetHttpPollingClientFactory pollingClientFactory =
             new JavaNetHttpPollingClientFactory(requestFactory);