1ESP-119_Http_headers_SourceSupport - Add Http header support for Http Lookup Source. (#21)

Signed-off-by: Krzysztof Chmielewski <krzysztof.chmielewski@getindata.com>

Signed-off-by: Krzysztof Chmielewski <krzysztof.chmielewski@getindata.com>

Author: kristoffSC

CHANGELOG.md

@@ -9,6 +9,7 @@
 - Add to Http Sink a new properties `gid.connector.http.sink.error.code` and `gid.connector.http.sink.error.code.exclude`
   to set HTTP status code that should be interpreted as errors.
 - Use Flink's format support to Http Lookup Source. 
+- Add HTTP Lookup source client header configuration via properties.
 
 ### Changed
 - Change dependency scope for `org.apache.flink.flink-connector-base` from `compile` to `provided`.

README.md

@@ -40,26 +40,30 @@ Flink SQL table definition:
 Enrichment Lookup Table
 ```roomsql
 CREATE TABLE Customers (
-	id STRING,
-	id2 STRING,
-	msg STRING,
-	uuid STRING,
-	details ROW<
-	  isActive BOOLEAN,
-	  nestedDetails ROW<
-	    balance STRING
-	  >
-	>
+    id STRING,
+    id2 STRING,
+    msg STRING,
+    uuid STRING,
+    details ROW<
+      isActive BOOLEAN,
+      nestedDetails ROW<
+        balance STRING
+      >
+    >
 ) WITH (
 'connector' = 'rest-lookup',
 'format' = 'json',
 'url' = 'http://localhost:8080/client',
 'asyncPolling' = 'true'
 )
 ```
+
 Data Source Table
 ```roomsql
-CREATE TABLE Orders (id STRING, id2 STRING, proc_time AS PROCTIME()
+CREATE TABLE Orders (
+    id STRING,
+    id2 STRING,
+    proc_time AS PROCTIME()
 ) WITH (
 'connector' = 'datagen',
 'rows-per-second' = '1',
@@ -70,8 +74,9 @@ CREATE TABLE Orders (id STRING, id2 STRING, proc_time AS PROCTIME()
 'fields.id2.start' = '2',
 'fields.id2.end' = '120'
 );
+```
 
-Using _Customers_ table in Flink SQL Lookup Join with Orders table:
+Using _Customers_ table in Flink SQL Lookup Join with _Orders_ table:
 
 ```roomsql
 SELECT o.id, o.id2, c.msg, c.uuid, c.isActive, c.balance FROM Orders AS o 
@@ -83,6 +88,31 @@ For Example:
 ``
 http://localhost:8080/client/service?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:
+`gid.connector.http.source.lookup.header.X-Content-Type-Options = nosniff`.
+
+Headers can be set using http lookup source table DDL. In example below, HTTP request done for `http-lookup` table will contain three headers:
+- `Origin`
+- `X-Content-Type-Options`
+- `Content-Type`
+
+```roomsql
+CREATE TABLE http-lookup (
+  id bigint,
+  some_field string
+) WITH (
+  'connector' = 'rest-lookup',
+  'format' = 'json',
+  'url' = 'http://localhost:8080/client',
+  'asyncPolling' = 'true',
+  'gid.connector.http.source.lookup.header.Origin' = '*',
+  'gid.connector.http.source.lookup.header.X-Content-Type-Options' = 'nosniff',
+  'gid.connector.http.source.lookup.header.Content-Type' = 'application/json'
+)
+```
+
 
 ### HTTP Sink
 The following example shows the minimum Table API example to create a [HttpDynamicSink](src/main/java/com/getindata/connectors/http/internal/table/HttpDynamicSink.java) that writes JSON values to an HTTP endpoint using POST method, assuming Flink has JAR of [JSON serializer](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/formats/json/) installed:
@@ -108,7 +138,7 @@ Due to the fact that `HttpSink` sends bytes inside HTTP request's body, one can
 
 Other examples of usage of the Table API can be found in [some tests](src/test/java/com/getindata/connectors/http/table/HttpDynamicSinkInsertTest.java).
 
-#### Http headers (currently supported only for HTTP Sink)
+#### Http headers
 It is possible to set HTTP headers that will be added to HTTP request send by sink connector.
 Headers are defined via property key `gid.connector.http.sink.header.HEADER_NAME = header value` for example:
 `gid.connector.http.sink.header.X-Content-Type-Options = nosniff`.

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

@@ -21,6 +21,8 @@ public interface SinkHttpClient {
      * @return the new {@link CompletableFuture} wrapping {@link SinkHttpClientResponse} that
      * completes when all requests have been sent and returned their statuses
      */
-    CompletableFuture<SinkHttpClientResponse> putRequests(List<HttpSinkRequestEntry> requestEntries,
-        String endpointUrl);
+    CompletableFuture<SinkHttpClientResponse> putRequests(
+        List<HttpSinkRequestEntry> requestEntries,
+        String endpointUrl
+    );
 }

src/main/java/com/getindata/connectors/http/internal/config/HttpConnectorConfigConstants.java

@@ -23,6 +23,9 @@ public final class HttpConnectorConfigConstants {
      */
     public static final String SINK_HEADER_PREFIX = GID_CONNECTOR_HTTP + "sink.header.";
 
+    public static final String LOOKUP_SOURCE_HEADER_PREFIX = GID_CONNECTOR_HTTP
+        + "source.lookup.header.";
+
     // Error code handling configuration.
     public static final String HTTP_ERROR_CODE_WHITE_LIST =
         GID_CONNECTOR_HTTP + "sink.error.code.exclude";

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

@@ -3,6 +3,7 @@
 import java.io.Serializable;
 import java.util.Collections;
 import java.util.List;
+import java.util.Properties;
 
 import lombok.Builder;
 import lombok.Data;
@@ -20,4 +21,7 @@ public class HttpLookupConfig implements Serializable {
 
     @Builder.Default
     private final boolean useAsync = false;
+
+    @Builder.Default
+    private final Properties properties = new Properties();
 }

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

@@ -2,16 +2,14 @@
 
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.HashSet;
 import java.util.List;
+import java.util.Properties;
 import java.util.Set;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.configuration.ConfigOptions;
-import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.table.api.DataTypes;
 import org.apache.flink.table.api.DataTypes.Field;
@@ -28,6 +26,8 @@
 import static org.apache.flink.table.types.utils.DataTypeUtils.removeTimeAttribute;
 
 import com.getindata.connectors.http.internal.PollingClientFactory;
+import com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants;
+import com.getindata.connectors.http.internal.utils.ConfigUtils;
 import static com.getindata.connectors.http.internal.table.lookup.HttpLookupConnectorOptions.ASYNC_POLLING;
 import static com.getindata.connectors.http.internal.table.lookup.HttpLookupConnectorOptions.URL;
 import static com.getindata.connectors.http.internal.table.lookup.HttpLookupConnectorOptions.URL_ARGS;
@@ -49,21 +49,20 @@ public static DataType row(List<Field> fields) {
 
     @Override
     public DynamicTableSource createDynamicTableSource(Context context) {
-        final FactoryUtil.TableFactoryHelper helper =
+        FactoryUtil.TableFactoryHelper helper =
             FactoryUtil.createTableFactoryHelper(this, context);
 
         ReadableConfig readableConfig = helper.getOptions();
-
-        validateOptions(context, readableConfig);
+        helper.validateExcept(HttpConnectorConfigConstants.GID_CONNECTOR_HTTP);
 
         DecodingFormat<DeserializationSchema<RowData>> decodingFormat =
             helper.discoverDecodingFormat(
                 DeserializationFormatFactory.class,
                 FactoryUtil.FORMAT
             );
 
-        PollingClientFactory<RowData> pollingClientFactory = new RestTablePollingClientFactory();
-        HttpLookupConfig lookupConfig = getHttpLookupOptions(readableConfig);
+        PollingClientFactory<RowData> pollingClientFactory = new JavaNetHttpPollingClientFactory();
+        HttpLookupConfig lookupConfig = getHttpLookupOptions(context, readableConfig);
 
         ResolvedSchema resolvedSchema = context.getCatalogTable().getResolvedSchema();
 
@@ -93,32 +92,19 @@ public Set<ConfigOption<?>> optionalOptions() {
         return Set.of(URL_ARGS, ASYNC_POLLING);
     }
 
-    private HttpLookupConfig getHttpLookupOptions(ReadableConfig config) {
+    private HttpLookupConfig getHttpLookupOptions(Context context, ReadableConfig config) {
+
+        Properties httpConnectorProperties =
+            ConfigUtils.getHttpConnectorProperties(context.getCatalogTable().getOptions());
+
         return HttpLookupConfig.builder()
             .url(config.get(URL))
             .arguments(convertArguments(config.get((URL_ARGS))))
             .useAsync(config.get(ASYNC_POLLING))
+            .properties(httpConnectorProperties)
             .build();
     }
 
-    // TODO FIX There is something ugly here. Verify with DataGenTableSourceFactory and refactor.
-    private void validateOptions(Context context, ReadableConfig readableConfig) {
-
-        Set<ConfigOption<?>> allOptions = new HashSet<>();
-        allOptions.addAll(optionalOptions());
-        allOptions.addAll(requiredOptions());
-        allOptions.add(ConfigOptions.key("connector").stringType().noDefaultValue());
-
-        FactoryUtil.validateFactoryOptions(requiredOptions(), allOptions, readableConfig);
-
-        Configuration options = new Configuration();
-        context.getCatalogTable().getOptions().forEach(options::setString);
-        Set<String> consumedOptionKeys = new HashSet<>();
-        allOptions.stream().map(ConfigOption::key).forEach(consumedOptionKeys::add);
-        FactoryUtil.validateUnconsumedKeys(factoryIdentifier(), options.keySet(),
-            consumedOptionKeys);
-    }
-
     private List<String> convertArguments(String arguments) {
 
         if (arguments == null || arguments.isEmpty() || arguments.isBlank()) {

src/main/java/com/getindata/connectors/http/internal/table/lookup/RestTablePollingClient.java renamed to src/main/java/com/getindata/connectors/http/internal/table/lookup/JavaNetHttpPollingClient.java

@@ -5,34 +5,58 @@
 import java.net.URISyntaxException;
 import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
+import java.net.http.HttpRequest.Builder;
 import java.net.http.HttpResponse;
 import java.net.http.HttpResponse.BodyHandlers;
 import java.time.Duration;
+import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
-import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 import org.apache.flink.table.data.RowData;
 
 import com.getindata.connectors.http.internal.PollingClient;
+import com.getindata.connectors.http.internal.utils.ConfigUtils;
 import com.getindata.connectors.http.internal.utils.uri.URIBuilder;
+import static com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants.LOOKUP_SOURCE_HEADER_PREFIX;
 
 /**
  * An implementation of {@link PollingClient} that uses Java 11's {@link HttpClient}.
  * This implementation supports HTTP traffic only.
  */
 @Slf4j
-@RequiredArgsConstructor
-public class RestTablePollingClient implements PollingClient<RowData> {
+public class JavaNetHttpPollingClient implements PollingClient<RowData> {
 
     private final HttpClient httpClient;
 
+    public JavaNetHttpPollingClient(
+            HttpClient httpClient,
+            DeserializationSchema<RowData> runtimeDecoder,
+            HttpLookupConfig options) {
+
+        this.httpClient = httpClient;
+        this.runtimeDecoder = runtimeDecoder;
+        this.options = options;
+
+        Map<String, String> headerMap = ConfigUtils.propertiesToMap(
+            options.getProperties(),
+            LOOKUP_SOURCE_HEADER_PREFIX,
+            String.class
+        );
+
+        this.headersAndValues = ConfigUtils.toHeaderAndValueArray(headerMap);
+    }
+
     private final DeserializationSchema<RowData> runtimeDecoder;
 
     private final HttpLookupConfig options;
 
+    private final String[] headersAndValues;
+
     @Override
     public Optional<RowData> pull(List<LookupArg> lookupArgs) {
         try {
@@ -45,14 +69,25 @@ public Optional<RowData> pull(List<LookupArg> lookupArgs) {
 
     // TODO Add Retry Policy And configure TimeOut from properties
     private Optional<RowData> queryAndProcess(List<LookupArg> params) throws Exception {
-        URI uri = buildUri(params);
-        HttpRequest request =
-            HttpRequest.newBuilder().uri(uri).GET().timeout(Duration.ofMinutes(2)).build();
-        HttpResponse<String> response = httpClient.send(request, BodyHandlers.ofString());
 
+        HttpRequest request = buildHttpRequest(params);
+        HttpResponse<String> response = httpClient.send(request, BodyHandlers.ofString());
         return processHttpResponse(response);
     }
 
+    private HttpRequest buildHttpRequest(List<LookupArg> params) throws URISyntaxException {
+        URI uri = buildUri(params);
+        Builder requestBuilder = HttpRequest.newBuilder()
+            .uri(uri).GET()
+            .timeout(Duration.ofMinutes(2));
+
+        if (headersAndValues.length != 0) {
+            requestBuilder.headers(headersAndValues);
+        }
+
+        return requestBuilder.build();
+    }
+
     private URI buildUri(List<LookupArg> params) throws URISyntaxException {
 
         URIBuilder uriBuilder = new URIBuilder(options.getUrl());
@@ -78,4 +113,9 @@ private Optional<RowData> processHttpResponse(HttpResponse<String> response)
             return Optional.empty();
         }
     }
+
+    @VisibleForTesting
+    String[] getHeadersAndValues() {
+        return Arrays.copyOf(headersAndValues, headersAndValues.length);
+    }
 }

src/main/java/com/getindata/connectors/http/internal/table/lookup/RestTablePollingClientFactory.java renamed to src/main/java/com/getindata/connectors/http/internal/table/lookup/JavaNetHttpPollingClientFactory.java

@@ -1,20 +1,25 @@
 package com.getindata.connectors.http.internal.table.lookup;
 
 import java.net.http.HttpClient;
+import java.net.http.HttpClient.Redirect;
 
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 import org.apache.flink.table.data.RowData;
 
 import com.getindata.connectors.http.internal.PollingClient;
 import com.getindata.connectors.http.internal.PollingClientFactory;
 
-public class RestTablePollingClientFactory implements PollingClientFactory<RowData> {
+public class JavaNetHttpPollingClientFactory implements PollingClientFactory<RowData> {
 
     @Override
     public PollingClient<RowData> createPollClient(
             HttpLookupConfig options,
             DeserializationSchema<RowData> schemaDecoder) {
 
-        return new RestTablePollingClient(HttpClient.newHttpClient(), schemaDecoder, options);
+        HttpClient httpClient = HttpClient.newBuilder()
+            .followRedirects(Redirect.NORMAL)
+            .build();
+
+        return new JavaNetHttpPollingClient(httpClient, schemaDecoder, options);
     }
 }

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

@@ -25,12 +25,15 @@ public class HttpDynamicTableSinkFactory extends AsyncDynamicTableSinkFactory {
     @Override
     public DynamicTableSink createDynamicTableSink(Context context) {
         final AsyncDynamicSinkContext factoryContext = new AsyncDynamicSinkContext(this, context);
+
+        // This is actually same as calling helper.getOptions();
         ReadableConfig tableOptions = factoryContext.getTableOptions();
 
         // Validate configuration
         FactoryUtil.createTableFactoryHelper(this, context)
             .validateExcept(HttpConnectorConfigConstants.GID_CONNECTOR_HTTP);
         validateHttpSinkOptions(tableOptions);
+
         Properties asyncSinkProperties =
             new AsyncSinkConfigurationValidator(tableOptions).getValidatedConfigurations();
 

src/test/java/com/getindata/connectors/http/internal/sink/httpclient/JavaNetSinkHttpClientTest.java

@@ -46,7 +46,7 @@ public void setUp() {
     }
 
     @Test
-    public void shouldBuildClientWithNoHeader() {
+    public void shouldBuildClientWithoutHeaders() {
 
         JavaNetSinkHttpClient client = new JavaNetSinkHttpClient(new Properties());
         assertThat(client.getHeadersAndValues()).isEmpty();