Merge pull request #29 from getindata/ESP-263_LookupSource_newRestMethods

ESP-263 - Adding suport for POST REST method to http Lookup Source

Author: kristoffSC

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## [Unreleased]
 
+### Added
+- Add support for other REST methods like PUT and POST to lookup source connector. The request method can be set using
+  new optional lookup-source property `lookup-method`. If property is not specified in table DDL, GET method will be used for
+  lookup queries.
+
 ## [0.5.0] - 2022-09-22
 
 ### Added

README.md

@@ -83,11 +83,21 @@ SELECT o.id, o.id2, c.msg, c.uuid, c.isActive, c.balance FROM Orders AS o
 JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c ON o.id = c.id AND o.id2 = c.id2
 ```
 
-The columns and their values used for JOIN `ON` condition will be used as HTTP get parameters where the column name will be used as a request parameter name. 
+The columns and their values used for JOIN `ON` condition will be used as HTTP GET parameters where the column name will be used as a request parameter name. 
+
 For Example: 
 ``
 http://localhost:8080/client/service?id=1&uuid=2
 ``
+
+Or for REST POST method they will be converted to Json and used as request body. In this case, json request body will look like this:
+```json
+{
+    "id": "1",
+    "uuid": "2"
+}
+```
+
 #### Http headers
 It is possible to set HTTP headers that will be added to HTTP request send by lookup source connector.
 Headers are defined via property key `gid.connector.http.source.lookup.header.HEADER_NAME = header value` for example:
@@ -256,6 +266,7 @@ If the used value starts from prefix `Basic `, it will be used as header value a
 | format                                                  | required | Flink's format name that should be used to decode REST response, Use `json` for a typical REST endpoint.                                             |
 | url                                                     | required | The base URL that should be use for GET requests. For example _http://localhost:8080/client_                                                         |
 | asyncPolling                                            | optional | true/false - determines whether Async Pooling should be used. Mechanism is based on Flink's Async I/O.                                               |
+| lookup-method                                           | optional | GET/POST/PUT (and any other) - determines what REST method should be used for lookup REST query. If not specified, `GET` method will be used.        |
 | gid.connector.http.lookup.error.code                    | optional | List of HTTP status codes that should be treated as errors by HTTP Source, separated with comma.                                                     |
 | gid.connector.http.lookup.error.code.exclude            | optional | List of HTTP status codes that should be excluded from the `gid.connector.http.lookup.error.code` list, separated with comma.                        |
 | gid.connector.http.security.cert.server                 | optional | Path to trusted HTTP server certificate that should be add to connectors key store. More than one path can be specified using `,` as path delimiter. |

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

@@ -11,6 +11,7 @@
  * {@link LookupQueryCreatorFactory}.
  */
 public interface LookupQueryCreator extends Serializable {
+
     /**
      * Create a lookup query (like the query appended to path in GET request)
      * out of the provided arguments.

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

@@ -30,6 +30,7 @@
  * }</pre>
  */
 public interface LookupQueryCreatorFactory extends Factory {
+
     /**
      * @return {@link LookupQueryCreator} custom lookup query creator instance
      */

src/main/java/com/getindata/connectors/http/internal/PollingClientFactory.java

@@ -4,14 +4,12 @@
 
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 
-import com.getindata.connectors.http.LookupQueryCreator;
 import com.getindata.connectors.http.internal.table.lookup.HttpLookupConfig;
 
 public interface PollingClientFactory<OUT> extends Serializable {
 
     PollingClient<OUT> createPollClient(
         HttpLookupConfig options,
-        DeserializationSchema<OUT> schemaDecoder,
-        LookupQueryCreator lookupQueryCreator
+        DeserializationSchema<OUT> schemaDecoder
     );
 }

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

@@ -0,0 +1,54 @@
+package com.getindata.connectors.http.internal.table.lookup;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.http.HttpRequest;
+import java.net.http.HttpRequest.BodyPublishers;
+import java.net.http.HttpRequest.Builder;
+import java.time.Duration;
+
+import com.getindata.connectors.http.LookupQueryCreator;
+import com.getindata.connectors.http.internal.HeaderPreprocessor;
+import com.getindata.connectors.http.internal.utils.uri.URIBuilder;
+
+/**
+ * Implementation of {@link HttpRequestFactory} for REST calls that sends their parameters using
+ * request body.
+ */
+public class BodyBasedRequestFactory extends RequestFactoryBase {
+
+    private final String methodName;
+
+    public BodyBasedRequestFactory(
+            String methodName,
+            LookupQueryCreator lookupQueryCreator,
+            HeaderPreprocessor headerPreprocessor,
+            HttpLookupConfig options) {
+
+        super(lookupQueryCreator, headerPreprocessor, options);
+        this.methodName = methodName.toUpperCase();
+    }
+
+    /**
+     * Method for preparing {@link HttpRequest.Builder} for REST request that sends their parameters
+     * in request body, for example PUT or POST methods
+     *
+     * @param lookupQuery lookup query used for request body.
+     * @return {@link HttpRequest.Builder} for given lookupQuery.
+     */
+    @Override
+    protected Builder setUpRequestMethod(String lookupQuery) {
+        return HttpRequest.newBuilder()
+            .uri(constructGetUri())
+            .method(methodName, BodyPublishers.ofString(lookupQuery))
+            .timeout(Duration.ofMinutes(2));
+    }
+
+    private URI constructGetUri() {
+        try {
+            return new URIBuilder(baseUrl).build();
+        } catch (URISyntaxException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

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

@@ -0,0 +1,50 @@
+package com.getindata.connectors.http.internal.table.lookup;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.http.HttpRequest;
+import java.net.http.HttpRequest.Builder;
+import java.time.Duration;
+
+import com.getindata.connectors.http.LookupQueryCreator;
+import com.getindata.connectors.http.internal.HeaderPreprocessor;
+import com.getindata.connectors.http.internal.utils.uri.URIBuilder;
+
+/**
+ * Implementation of {@link HttpRequestFactory} for GET REST calls.
+ */
+public class GetRequestFactory extends RequestFactoryBase {
+
+    public GetRequestFactory(
+            LookupQueryCreator lookupQueryCreator,
+            HeaderPreprocessor headerPreprocessor,
+            HttpLookupConfig options) {
+
+        super(lookupQueryCreator, headerPreprocessor, options);
+    }
+
+    /**
+     * Method for preparing {@link HttpRequest.Builder} for REST GET request, where lookupQuery
+     * is used as query parameters for example:
+     * <pre>
+     *     http:localhost:8080/service?id=1
+     * </pre>
+     * @param lookupQuery lookup query used for request query parameters.
+     * @return {@link HttpRequest.Builder} for given GET lookupQuery
+     */
+    @Override
+    protected Builder setUpRequestMethod(String lookupQuery) {
+        return HttpRequest.newBuilder()
+            .uri(constructGetUri(lookupQuery))
+            .GET()
+            .timeout(Duration.ofMinutes(2));
+    }
+
+    private URI constructGetUri(String lookupQuery) {
+        try {
+            return new URIBuilder(baseUrl + "?" + lookupQuery).build();
+        } catch (URISyntaxException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

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

@@ -3,7 +3,6 @@
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.ConfigOptions;
 
-import com.getindata.connectors.http.internal.table.lookup.querycreators.GenericGetQueryCreatorFactory;
 import static com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants.SOURCE_LOOKUP_QUERY_CREATOR_IDENTIFIER;
 
 public class HttpLookupConnectorOptions {
@@ -26,8 +25,14 @@ public class HttpLookupConnectorOptions {
             .defaultValue(false)
             .withDescription("Whether to use Sync and Async polling mechanism");
 
+    public static final ConfigOption<String> LOOKUP_METHOD =
+        ConfigOptions.key("lookup-method")
+            .stringType()
+            .defaultValue("GET")
+            .withDescription("Method used for REST executed by lookup connector.");
+
     public static final ConfigOption<String> LOOKUP_QUERY_CREATOR_IDENTIFIER =
         ConfigOptions.key(SOURCE_LOOKUP_QUERY_CREATOR_IDENTIFIER)
             .stringType()
-            .defaultValue(GenericGetQueryCreatorFactory.IDENTIFIER);
+            .noDefaultValue();
 }

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

@@ -15,7 +15,6 @@
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.DataType;
 
-import com.getindata.connectors.http.LookupQueryCreator;
 import com.getindata.connectors.http.internal.PollingClientFactory;
 import com.getindata.connectors.http.internal.table.lookup.HttpTableLookupFunction.ColumnData;
 
@@ -32,8 +31,6 @@ public class HttpLookupTableSource
 
     private final DecodingFormat<DeserializationSchema<RowData>> decodingFormat;
 
-    private final LookupQueryCreator lookupQueryCreator;
-
     @Override
     public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
         String[] keyNames = new String[context.getKeys().length];
@@ -53,8 +50,7 @@ public DynamicTableSource copy() {
             physicalRowDataType,
             pollingClientFactory,
             lookupConfig,
-            decodingFormat,
-            lookupQueryCreator
+            decodingFormat
         );
     }
 
@@ -84,7 +80,6 @@ private LookupRuntimeProvider buildLookupFunction(String[] keyNames, LookupConte
                 .pollingClientFactory(pollingClientFactory)
                 .schemaDecoder(schemaDecoder)
                 .columnData(columnData)
-                .lookupQueryCreator(lookupQueryCreator)
                 .options(lookupConfig)
                 .build();
 

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

@@ -26,9 +26,13 @@
 import static org.apache.flink.table.types.utils.DataTypeUtils.removeTimeAttribute;
 
 import com.getindata.connectors.http.LookupQueryCreatorFactory;
+import com.getindata.connectors.http.internal.HeaderPreprocessor;
 import com.getindata.connectors.http.internal.PollingClientFactory;
 import com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants;
+import com.getindata.connectors.http.internal.table.lookup.querycreators.GenericGetQueryCreatorFactory;
+import com.getindata.connectors.http.internal.table.lookup.querycreators.GenericJsonQueryCreatorFactory;
 import com.getindata.connectors.http.internal.utils.ConfigUtils;
+import com.getindata.connectors.http.internal.utils.HttpHeaderUtils;
 import static com.getindata.connectors.http.internal.table.lookup.HttpLookupConnectorOptions.*;
 
 public class HttpLookupTableSourceFactory implements DynamicTableSourceFactory {
@@ -60,16 +64,26 @@ public DynamicTableSource createDynamicTableSource(Context context) {
                 FactoryUtil.FORMAT
             );
 
-        final LookupQueryCreatorFactory lookupQueryCreatorFactory =
+        String lookupMethod = readableConfig.get(LOOKUP_METHOD);
+
+        LookupQueryCreatorFactory lookupQueryCreatorFactory =
             FactoryUtil.discoverFactory(
                 context.getClassLoader(),
                 LookupQueryCreatorFactory.class,
-                readableConfig.get(LOOKUP_QUERY_CREATOR_IDENTIFIER)
+                readableConfig.getOptional(LOOKUP_QUERY_CREATOR_IDENTIFIER).orElse(
+                    (lookupMethod.equalsIgnoreCase("GET") ?
+                        GenericGetQueryCreatorFactory.IDENTIFIER :
+                        GenericJsonQueryCreatorFactory.IDENTIFIER)
+                )
             );
 
-        PollingClientFactory<RowData> pollingClientFactory = new JavaNetHttpPollingClientFactory();
         HttpLookupConfig lookupConfig = getHttpLookupOptions(context, readableConfig);
 
+        // TODO Consider this to be injected as method argument or factory field
+        //  so user could set this using API.
+        PollingClientFactory<RowData> pollingClientFactory =
+            createPollingClientFactory(lookupMethod, lookupQueryCreatorFactory, lookupConfig);
+
         ResolvedSchema resolvedSchema = context.getCatalogTable().getResolvedSchema();
 
         DataType physicalRowDataType =
@@ -79,8 +93,7 @@ public DynamicTableSource createDynamicTableSource(Context context) {
             physicalRowDataType,
             pollingClientFactory,
             lookupConfig,
-            decodingFormat,
-            lookupQueryCreatorFactory.createLookupQueryCreator()
+            decodingFormat
         );
     }
 
@@ -96,7 +109,29 @@ public Set<ConfigOption<?>> requiredOptions() {
 
     @Override
     public Set<ConfigOption<?>> optionalOptions() {
-        return Set.of(URL_ARGS, ASYNC_POLLING);
+        return Set.of(URL_ARGS, ASYNC_POLLING, LOOKUP_METHOD);
+    }
+
+    private PollingClientFactory<RowData> createPollingClientFactory(
+        String lookupMethod,
+        LookupQueryCreatorFactory lookupQueryCreatorFactory,
+        HttpLookupConfig lookupConfig) {
+
+        HeaderPreprocessor headerPreprocessor = HttpHeaderUtils.createDefaultHeaderPreprocessor();
+
+        HttpRequestFactory requestFactory = (lookupMethod.equalsIgnoreCase("GET")) ?
+            new GetRequestFactory(
+                lookupQueryCreatorFactory.createLookupQueryCreator(),
+                headerPreprocessor,
+                lookupConfig) :
+            new BodyBasedRequestFactory(
+                lookupMethod,
+                lookupQueryCreatorFactory.createLookupQueryCreator(),
+                headerPreprocessor,
+                lookupConfig
+            );
+
+        return new JavaNetHttpPollingClientFactory(requestFactory);
     }
 
     private HttpLookupConfig getHttpLookupOptions(Context context, ReadableConfig config) {

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

@@ -0,0 +1,21 @@
+package com.getindata.connectors.http.internal.table.lookup;
+
+import java.io.Serializable;
+import java.net.http.HttpRequest;
+import java.util.List;
+
+import com.getindata.connectors.http.LookupArg;
+
+/**
+ * Factory for creating {@link HttpRequest} objects for Rest clients.
+ */
+public interface HttpRequestFactory extends Serializable {
+
+    /**
+     * Creates {@link HttpRequest} from given List of {@link LookupArg} objects.
+     *
+     * @param params {@link LookupArg} objects used for building http request.
+     * @return {@link HttpRequest} created from {@link LookupArg}
+     */
+    HttpRequest buildLookupRequest(List<LookupArg> params);
+}

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

@@ -20,7 +20,6 @@
 import org.apache.flink.table.functions.TableFunction;
 
 import com.getindata.connectors.http.LookupArg;
-import com.getindata.connectors.http.LookupQueryCreator;
 import com.getindata.connectors.http.internal.PollingClient;
 import com.getindata.connectors.http.internal.PollingClientFactory;
 
@@ -31,8 +30,6 @@ public class HttpTableLookupFunction extends TableFunction<RowData> {
 
     private final DeserializationSchema<RowData> schemaDecoder;
 
-    private final LookupQueryCreator lookupQueryCreator;
-
     @Getter
     private final ColumnData columnData;
 
@@ -48,22 +45,20 @@ private HttpTableLookupFunction(
         PollingClientFactory<RowData> pollingClientFactory,
         DeserializationSchema<RowData> schemaDecoder,
         ColumnData columnData,
-        HttpLookupConfig options,
-        LookupQueryCreator lookupQueryCreator) {
+        HttpLookupConfig options) {
 
         this.pollingClientFactory = pollingClientFactory;
         this.schemaDecoder = schemaDecoder;
         this.columnData = columnData;
         this.options = options;
-        this.lookupQueryCreator = lookupQueryCreator;
     }
 
     @Override
     public void open(FunctionContext context) throws Exception {
         super.open(context);
         this.localHttpCallCounter = new AtomicInteger(0);
         this.client = pollingClientFactory
-            .createPollClient(options, schemaDecoder, lookupQueryCreator);
+            .createPollClient(options, schemaDecoder);
 
         context
             .getMetricGroup()