Issue54 Fix class loader and Prefixed Config option issues (#68)

* Issue54 early prototype not working

Signed-off-by: David Radley <[email protected]>

* fixed the class loader part

Signed-off-by: David Radley <[email protected]>

* Change Prefixed Config Option to not extend  ConfigOption

Signed-off-by: David Radley <[email protected]>

* Update Change log

Signed-off-by: David Radley <[email protected]>

* update doc link to 1.16

Signed-off-by: David Radley <[email protected]>

* update more links to 1.16

Signed-off-by: David Radley <[email protected]>

* update more links to 1.16

Signed-off-by: David Radley <[email protected]>

* Use reflection to update the key in the ConfigOption

Signed-off-by: David Radley <[email protected]>

* Re-implement reflection after review feedback

Signed-off-by: David Radley <[email protected]>

* Remove unused classloader field

Signed-off-by: David Radley <[email protected]>

* fix compile error and address review feedback

Signed-off-by: David Radley <[email protected]>

---------

Signed-off-by: David Radley <[email protected]>

Author: davidradl

CHANGELOG.md

@@ -4,6 +4,19 @@
 
 ## [0.10.0] - 2023-07-05
 
+
+### Fixed 
+
+    Fixed an issue where SQL Client did not work with the connector at Flink 1.16.
+    
+    This required a change to use a different classloader in the lookup join processing. 
+    As well as the classloader change, a change to the PrefixedConfigOption implementation was
+    required, because it was implemented as an extension to ConfigOption; which produced  
+    access errors when trying to access the parent class protected methods (the parent class was loaded 
+    using a different classloader). The new implementation is not an extension; instead it holds an
+    instance of the ConfigOption as a private variable and uses reflection to instantiate a cloned 
+    ConfigOption object with the prefixed key. 
+
 ### Added
 
 -   Add support for batch request submission in HTTP sink. The mode can be changed by setting

README.md

@@ -19,7 +19,7 @@ In case of updating http-connector please see [Breaking changes](#breaking-chang
 ## Prerequisites
 * Java 11
 * Maven 3
-* Flink 1.15+
+* Flink 1.16+
 
 ## Runtime dependencies
 This connector has few Flink's runtime dependencies, that are expected to be provided.
@@ -439,16 +439,15 @@ Project build command: `mvn package`. </br>
 Detailed test report can be found under `target/site/jacoco/index.xml`.
 
 ## Demo application
-**Note**: This demo works only for Flink-1.15.x even though connector can be used with Flink 1.16.
-This problem is addressed by https://github.com/getindata/flink-http-connector/issues/54 and will be fix in next relesees. 
+**Note**: This demo works only for Flink-1.15x.
 
 You can test this connector using simple mock http server provided with this repository and Flink SQL-client. 
 The mock server can be started from IDE (currently only this way) by running `HttpStubApp::main` method. 
 It will start HTTP server listening on `http://localhost:8080/client`
 
 Steps to follow:
 - Run Mock HTTP server from `HttpStubApp::main` method.
-- Start your Flink cluster, for example as described under https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/try-flink/local_installation/
+- Start your Flink cluster, for example as described under https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/try-flink/local_installation/
 - Start Flink SQL Client [6] by calling: `./bin/sql-client.sh -j flink-http-connector-1.0-SNAPSHOT.jar`
 - Execute SQL statements:
 Create Data Stream source Table:

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

@@ -3,6 +3,7 @@
 import java.io.Serializable;
 
 import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.factories.DynamicTableFactory;
 import org.apache.flink.table.factories.Factory;
 
 import com.getindata.connectors.http.internal.table.lookup.HttpLookupTableSource;
@@ -40,5 +41,6 @@ public interface LookupQueryCreatorFactory extends Factory, Serializable {
      */
     LookupQueryCreator createLookupQueryCreator(
         ReadableConfig readableConfig,
-        LookupRow lookupRow);
+        LookupRow lookupRow,
+        DynamicTableFactory.Context dynamicTableFactoryContext);
 }

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

@@ -5,6 +5,7 @@
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.table.api.DataTypes;
 import org.apache.flink.table.api.DataTypes.Field;
 import org.apache.flink.table.connector.format.DecodingFormat;
@@ -15,6 +16,7 @@
 import org.apache.flink.table.connector.source.abilities.SupportsLimitPushDown;
 import org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown;
 import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableFactory;
 import org.apache.flink.table.factories.FactoryUtil;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.LogicalType;
@@ -40,29 +42,33 @@ public class HttpLookupTableSource
 
     private final HttpLookupConfig lookupConfig;
 
+    private final DynamicTableFactory.Context dynamicTableFactoryContext;
+
     private final DecodingFormat<DeserializationSchema<RowData>> decodingFormat;
 
     public HttpLookupTableSource(
             DataType physicalRowDataType,
             HttpLookupConfig lookupConfig,
-            DecodingFormat<DeserializationSchema<RowData>> decodingFormat) {
+            DecodingFormat<DeserializationSchema<RowData>> decodingFormat,
+            DynamicTableFactory.Context dynamicTablecontext) {
 
         this.physicalRowDataType = physicalRowDataType;
         this.lookupConfig = lookupConfig;
         this.decodingFormat = decodingFormat;
+        this.dynamicTableFactoryContext = dynamicTablecontext;
     }
 
     @Override
-    public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
+    public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext lookupContext) {
 
-        LookupRow lookupRow = extractLookupRow(context.getKeys());
+        LookupRow lookupRow = extractLookupRow(lookupContext.getKeys());
 
         DeserializationSchema<RowData> responseSchemaDecoder =
-            decodingFormat.createRuntimeDecoder(context, physicalRowDataType);
+            decodingFormat.createRuntimeDecoder(lookupContext, physicalRowDataType);
 
         LookupQueryCreatorFactory lookupQueryCreatorFactory =
             FactoryUtil.discoverFactory(
-                Thread.currentThread().getContextClassLoader(),
+                this.dynamicTableFactoryContext.getClassLoader(),
                 LookupQueryCreatorFactory.class,
                 lookupConfig.getReadableConfig().getOptional(LOOKUP_QUERY_CREATOR_IDENTIFIER)
                     .orElse(
@@ -71,11 +77,12 @@ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
                             GenericJsonQueryCreatorFactory.IDENTIFIER)
                     )
             );
-
+        ReadableConfig readableConfig = lookupConfig.getReadableConfig();
         LookupQueryCreator lookupQueryCreator =
             lookupQueryCreatorFactory.createLookupQueryCreator(
-                lookupConfig.getReadableConfig(),
-                lookupRow
+                readableConfig,
+                lookupRow,
+                dynamicTableFactoryContext
             );
 
         PollingClientFactory<RowData> pollingClientFactory =
@@ -104,7 +111,8 @@ public DynamicTableSource copy() {
         return new HttpLookupTableSource(
             physicalRowDataType,
             lookupConfig,
-            decodingFormat
+            decodingFormat,
+            dynamicTableFactoryContext
         );
     }
 

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

@@ -43,9 +43,9 @@ public static DataType row(List<Field> fields) {
     }
 
     @Override
-    public DynamicTableSource createDynamicTableSource(Context context) {
+    public DynamicTableSource createDynamicTableSource(Context dynamicTableContext) {
         FactoryUtil.TableFactoryHelper helper =
-            FactoryUtil.createTableFactoryHelper(this, context);
+            FactoryUtil.createTableFactoryHelper(this, dynamicTableContext);
 
         ReadableConfig readableConfig = helper.getOptions();
         helper.validateExcept(
@@ -61,17 +61,18 @@ public DynamicTableSource createDynamicTableSource(Context context) {
                 FactoryUtil.FORMAT
             );
 
-        HttpLookupConfig lookupConfig = getHttpLookupOptions(context, readableConfig);
+        HttpLookupConfig lookupConfig = getHttpLookupOptions(dynamicTableContext, readableConfig);
 
-        ResolvedSchema resolvedSchema = context.getCatalogTable().getResolvedSchema();
+        ResolvedSchema resolvedSchema = dynamicTableContext.getCatalogTable().getResolvedSchema();
 
         DataType physicalRowDataType =
             toRowDataType(resolvedSchema.getColumns(), Column::isPhysical);
 
         return new HttpLookupTableSource(
             physicalRowDataType,
             lookupConfig,
-            decodingFormat
+            decodingFormat,
+            dynamicTableContext
         );
     }
 

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

@@ -4,11 +4,13 @@
 
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.factories.DynamicTableFactory;
 
 import com.getindata.connectors.http.LookupQueryCreator;
 import com.getindata.connectors.http.LookupQueryCreatorFactory;
 import com.getindata.connectors.http.internal.table.lookup.LookupRow;
 
+
 /**
  * Factory for creating {@link ElasticSearchLiteQueryCreator}.
  */
@@ -19,7 +21,9 @@ public class ElasticSearchLiteQueryCreatorFactory implements LookupQueryCreatorF
     @Override
     public LookupQueryCreator createLookupQueryCreator(
             ReadableConfig readableConfig,
-            LookupRow lookupRow) {
+            LookupRow lookupRow,
+            DynamicTableFactory.Context dynamicTableFactoryContext
+    ) {
         return new ElasticSearchLiteQueryCreator(lookupRow);
     }
 

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

@@ -4,11 +4,13 @@
 
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.factories.DynamicTableFactory;
 
 import com.getindata.connectors.http.LookupQueryCreator;
 import com.getindata.connectors.http.LookupQueryCreatorFactory;
 import com.getindata.connectors.http.internal.table.lookup.LookupRow;
 
+
 /**
  * Factory for creating {@link GenericGetQueryCreator}.
  */
@@ -19,7 +21,8 @@ public class GenericGetQueryCreatorFactory implements LookupQueryCreatorFactory
     @Override
     public LookupQueryCreator createLookupQueryCreator(
             ReadableConfig readableConfig,
-            LookupRow lookupRow) {
+            LookupRow lookupRow,
+            DynamicTableFactory.Context dynamicTableFactoryContext) {
         return new GenericGetQueryCreator(lookupRow);
     }
 

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

@@ -8,6 +8,7 @@
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.table.connector.format.EncodingFormat;
 import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableFactory;
 import org.apache.flink.table.factories.FactoryUtil;
 import org.apache.flink.table.factories.SerializationFormatFactory;
 
@@ -26,22 +27,24 @@ public class GenericJsonQueryCreatorFactory implements LookupQueryCreatorFactory
     @Override
     public LookupQueryCreator createLookupQueryCreator(
             ReadableConfig readableConfig,
-            LookupRow lookupRow) {
+            LookupRow lookupRow,
+            DynamicTableFactory.Context dynamicTableFactoryContext) {
 
         String formatIdentifier = readableConfig.get(LOOKUP_REQUEST_FORMAT);
         SerializationFormatFactory jsonFormatFactory =
             FactoryUtil.discoverFactory(
-                Thread.currentThread().getContextClassLoader(),
+                dynamicTableFactoryContext.getClassLoader(),
                 SerializationFormatFactory.class,
                 formatIdentifier
             );
-
-        EncodingFormat<SerializationSchema<RowData>>
-            encoder = jsonFormatFactory.createEncodingFormat(
-            null,
+        QueryFormatAwareConfiguration queryFormatAwareConfiguration =
             new QueryFormatAwareConfiguration(
                 LOOKUP_REQUEST_FORMAT.key() + "." + formatIdentifier,
-                (Configuration) readableConfig)
+                (Configuration) readableConfig);
+        EncodingFormat<SerializationSchema<RowData>>
+            encoder = jsonFormatFactory.createEncodingFormat(
+            dynamicTableFactoryContext,
+            queryFormatAwareConfiguration
         );
 
         SerializationSchema<RowData> serializationSchema =

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

@@ -0,0 +1,85 @@
+package com.getindata.connectors.http.internal.table.lookup.querycreators;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.FallbackKey;
+
+/**
+ * This is a ConfigOption that has an associated config option and prefix.
+ *
+ * Note that this Class used to extend ConfigOption,
+ * but at Flink 1.16, there was a new way of doing class loaders
+ * for custom content, so we could no longer extend ConfigOption.
+ */
+public class PrefixedConfigOption<T> {
+    /**
+     * configOption to decorate
+     */
+    private ConfigOption configOption;
+
+    public ConfigOption getConfigOption() {
+        return configOption;
+    }
+
+    /**
+     * This constructor creates a new clone of the supplied option 'other' with
+     * the prefix prefixing the key. We create a new object, because we do
+     * not want to mutate a Flink object that we did not create.
+     *
+     * @param keyPrefix prefix that will be added to decorate the {@link ConfigOption} key.
+     * @param other  original {@link ConfigOption} to clone and decorate.
+     */
+    public PrefixedConfigOption(String keyPrefix, ConfigOption<T> other) {
+        String prefixedKey = keyPrefix + other.key();
+        Class clazz;
+        boolean isList;
+
+        try {
+            // get clazz
+            Field field = other.getClass().getDeclaredField("clazz");
+            field.setAccessible(true);
+            clazz = (Class) field.get(other);
+
+            // get isList
+            field = other.getClass().getDeclaredField("isList");
+            field.setAccessible(true);
+            isList = (Boolean) field.get(other);
+
+            /*
+            * Create a new ConfigOption based on other, but with a prefixed key.
+            * At 1.16 we cannot access the protected fields / constructor in the supplied
+            * configOption as this object is loaded using a different classloader.
+            * Without changing Flink to make the constructor, methods and fields public, we need
+            * to use reflection to access and create the new prefixed ConfigOption. It is not
+            * great practise to use reflection, but getting round this classloader issue
+            * necessitates it's use.
+            */
+            Constructor constructor = other.getClass().getDeclaredConstructors()[0];
+            constructor.setAccessible(true);
+            configOption = (ConfigOption) constructor.newInstance(prefixedKey,
+                    clazz,
+                    other.description(),
+                    other.defaultValue(),
+                    isList,
+                    getFallbackKeys(other));
+        } catch (InstantiationException |
+                 IllegalAccessException |
+                 InvocationTargetException |
+                 NoSuchFieldException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static FallbackKey[] getFallbackKeys(ConfigOption<?> other) {
+        List<FallbackKey> fallbackKeys = new ArrayList<>();
+        for (FallbackKey fallbackKey : other.fallbackKeys()) {
+            fallbackKeys.add(fallbackKey);
+        }
+        return fallbackKeys.toArray(new FallbackKey[0]);
+    }
+}

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

@@ -4,7 +4,6 @@
 
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.PrefixedConfigOption;
 import org.apache.flink.table.factories.SerializationFormatFactory;
 
 /**
@@ -36,8 +35,8 @@ class QueryFormatAwareConfiguration extends Configuration {
      */
     @Override
     public <T> Optional<T> getOptional(ConfigOption<T> option) {
-        PrefixedConfigOption<T> configOption = new PrefixedConfigOption<>(queryFormatName, option);
-        return super.getOptional(configOption);
+        PrefixedConfigOption<T> prefixedConfigOption =
+              new PrefixedConfigOption<>(queryFormatName, option);
+        return super.getOptional(prefixedConfigOption.getConfigOption());
     }
-
 }