HTTP-179 Allow HTTP version to be specified

davidradl · davidradl · commit 4b3d7fb51924 · 2025-11-04T10:56:37.000Z
Signed-off-by: davidradl &lt;david_radley@uk.ibm.com&gt;
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,7 +2,9 @@
 
 ## [Unreleased]
 
-- Added http 2 support.
+- Ability to specify http versions for http lookups. 
+- Amend to not log HTTP request response and header values by default.
+
 
 ## [0.22.0] - 2025-10-03
 
diff --git a/README.md b/README.md
@@ -9,10 +9,8 @@ The HTTP TableLookup connector that allows for pulling data from external system
 Please use [releases](https://github.com/getindata/flink-http-connector/releases) instead of the `main` branch in order to get a stable set of binaries.
 
 The goal for HTTP TableLookup connector was to use it in Flink SQL statement as a standard table that can be later joined with other stream using pure SQL Flink.
- 
-Currently, HTTP source connector supports only Lookup Joins (TableLookup) [1] in Table/SQL API.
-`HttpSink` supports both Streaming API (when using [HttpSink](src/main/java/com/getindata/connectors/http/internal/sink/HttpSink.java) built using [HttpSinkBuilder](src/main/java/com/getindata/connectors/http/internal/sink/HttpSinkBuilder.java)) and the Table API (using connector created in [HttpDynamicTableSinkFactory](src/main/java/com/getindata/connectors/http/internal/table/HttpDynamicTableSinkFactory.java)). 
-Note that the connector will work with both http 1.1 and http 2 endpoints.
+
+`HttpSink` supports both Streaming API (when using [HttpSink](src/main/java/com/getindata/connectors/http/internal/sink/HttpSink.java) built using [HttpSinkBuilder](src/main/java/com/getindata/connectors/http/internal/sink/HttpSinkBuilder.java)) and the Table API (using connector created in [HttpDynamicTableSinkFactory](src/main/java/com/getindata/connectors/http/internal/table/HttpDynamicTableSinkFactory.java)).
 
 ## Updating the connector
 In case of updating http-connector please see [Breaking changes](#breaking-changes) section.
@@ -428,10 +426,9 @@ that implements interface `HttpPostRequestCallbackFactory<HttpRequest>` to creat
 of class `CustomHttpSinkPostRequestCallbackFactory` in `resources/META-INF/services/org.apache.flink.table.factories.Factory` file 
 and then reference identifier `rest-sink-logger` in the HttpSink DDL property field `gid.connector.http.sink.request-callback`.
 
-  A default implementation that logs those pairs as *INFO* level logs using Slf4j
-([Slf4jHttpPostRequestCallback](src/main/java/com/getindata/connectors/http/internal/table/sink/Slf4jHttpPostRequestCallback.java))
-is provided.
-
+A default implementation that logs those pairs as *INFO* level logs using Slf4j ([Slf4jHttpPostRequestCallback](src/main/java/com/getindata/connectors/http/internal/table/sink/Slf4jHttpPostRequestCallback.java)) is provided.
+If you would like to log more http content (that maybe contain sensitive information), then you can provide a customized version
+of this callback; for inspiration on how to customize in this way, look back in the git history of this file.
 
 - Http Lookup Source processes responses that it gets from the HTTP endpoint along their respective requests. One can customize the
 behaviour of the additional stage of processing done by Table Function API by implementing
@@ -582,6 +579,7 @@ be requested if the current time is later than the cached token expiry time minu
 | gid.connector.http.security.oidc.token.endpoint.url                                  | optional | OIDC `Token Endpoint` url, to which the token request will be issued                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             |
 | gid.connector.http.security.oidc.token.expiry.reduction                              | optional | OIDC tokens will be requested if the current time is later than the cached token expiry time minus this value.                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
 | gid.connector.http.source.lookup.request.timeout                                     | optional | Sets HTTP request timeout in seconds. If not specified, the default value of 30 seconds will be used.                                                                                                                                                                                                                                                                                                                                                                                                                                                            |
+| gid.connector.http.source.lookup.http-version                                        | optional | Version of HTTP to use for lookup http requests. The valid values are HTTP_1_1 and HTTP_2, which specify HTTP 1.1 or 2 respectively. This option may be required as HTTP_1_1, if the endpoint is HTTP 1.1, because some http 1.1 endpoints reject HTTP Version 2 calls, with 'Invalid HTTP request received' and 'HTTP/2 upgrade not supported'.                                                                                                                                                                                                                 |
 | gid.connector.http.source.lookup.request.thread-pool.size                            | optional | Sets the size of pool thread for HTTP lookup request processing. Increasing this value would mean that more concurrent requests can be processed in the same time. If not specified, the default value of 8 threads will be used.                                                                                                                                                                                                                                                                                                                                |
 | gid.connector.http.source.lookup.response.thread-pool.size                           | optional | Sets the size of pool thread for HTTP lookup response processing. Increasing this value would mean that more concurrent requests can be processed in the same time. If not specified, the default value of 4 threads will be used.                                                                                                                                                                                                                                                                                                                               |
 | gid.connector.http.source.lookup.use-raw-authorization-header                        | optional | If set to `'true'`, uses the raw value set for the `Authorization` header, without transformation for Basic Authentication (base64, addition of "Basic " prefix). If not specified, defaults to `'false'`.                                                                                                                                                                                                                                                                                                                                                       |
diff --git a/src/main/java/com/getindata/connectors/http/internal/config/HttpConnectorConfigConstants.java b/src/main/java/com/getindata/connectors/http/internal/config/HttpConnectorConfigConstants.java
@@ -59,6 +59,9 @@ public final class HttpConnectorConfigConstants {
     public static final String SOURCE_LOOKUP_QUERY_CREATOR_IDENTIFIER =
         SOURCE_LOOKUP_PREFIX + "query-creator";
 
+    public static final String SOURCE_LOOKUP_QUERY_HTTP_VERSION =
+        SOURCE_LOOKUP_PREFIX + "http-version";
+
     // -------------- HTTPS security settings --------------
     public static final String ALLOW_SELF_SIGNED =
         GID_CONNECTOR_HTTP + "security.cert.server.allowSelfSigned";
diff --git a/src/main/java/com/getindata/connectors/http/internal/sink/httpclient/BatchRequestSubmitter.java b/src/main/java/com/getindata/connectors/http/internal/sink/httpclient/BatchRequestSubmitter.java
@@ -124,7 +124,7 @@ private HttpRequest buildHttpRequest(List<HttpSinkRequestEntry> reqeustBatch, UR
             Builder requestBuilder = java.net.http.HttpRequest
                 .newBuilder()
                 .uri(endpointUri)
-                .version(Version.HTTP_2)
+                .version(Version.HTTP_1_1) // consider allowing users to specify http 2 to override
                 .timeout(Duration.ofSeconds(httpRequestTimeOutSeconds))
                 .method(method, publisher);
 
diff --git a/src/main/java/com/getindata/connectors/http/internal/sink/httpclient/PerRequestSubmitter.java b/src/main/java/com/getindata/connectors/http/internal/sink/httpclient/PerRequestSubmitter.java
@@ -62,7 +62,7 @@ private HttpRequest buildHttpRequest(HttpSinkRequestEntry requestEntry, URI endp
         Builder requestBuilder = java.net.http.HttpRequest
             .newBuilder()
             .uri(endpointUri)
-            .version(Version.HTTP_2)
+            .version(Version.HTTP_1_1) // consider allowing users to specify http 2 to override
             .timeout(Duration.ofSeconds(httpRequestTimeOutSeconds))
             .method(requestEntry.method,
                 BodyPublishers.ofByteArray(requestEntry.element));
diff --git a/src/main/java/com/getindata/connectors/http/internal/table/lookup/BodyBasedRequestFactory.java b/src/main/java/com/getindata/connectors/http/internal/table/lookup/BodyBasedRequestFactory.java
@@ -5,7 +5,6 @@
 import java.net.http.HttpRequest;
 import java.net.http.HttpRequest.BodyPublishers;
 import java.net.http.HttpRequest.Builder;
-import java.time.Duration;
 
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
@@ -42,10 +41,11 @@ public BodyBasedRequestFactory(
      */
     @Override
     protected Builder setUpRequestMethod(LookupQueryInfo lookupQueryInfo) {
-        return HttpRequest.newBuilder()
+        HttpRequest.Builder builder = super.setUpRequestMethod(lookupQueryInfo);
+        builder
             .uri(constructUri(lookupQueryInfo))
-            .method(methodName, BodyPublishers.ofString(lookupQueryInfo.getLookupQuery()))
-            .timeout(Duration.ofSeconds(this.httpRequestTimeOutSeconds));
+            .method(methodName, BodyPublishers.ofString(lookupQueryInfo.getLookupQuery()));
+        return builder;
     }
 
     @Override
diff --git a/src/main/java/com/getindata/connectors/http/internal/table/lookup/GetRequestFactory.java b/src/main/java/com/getindata/connectors/http/internal/table/lookup/GetRequestFactory.java
@@ -4,7 +4,6 @@
 import java.net.URISyntaxException;
 import java.net.http.HttpRequest;
 import java.net.http.HttpRequest.Builder;
-import java.time.Duration;
 
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
@@ -48,10 +47,11 @@ protected Logger getLogger() {
      */
     @Override
     protected Builder setUpRequestMethod(LookupQueryInfo lookupQueryInfo) {
-        return HttpRequest.newBuilder()
+        HttpRequest.Builder builder = super.setUpRequestMethod(lookupQueryInfo);
+        builder
             .uri(constructGetUri(lookupQueryInfo))
-            .GET()
-            .timeout(Duration.ofSeconds(this.httpRequestTimeOutSeconds));
+            .GET();
+        return builder;
     }
 
     URI constructGetUri(LookupQueryInfo lookupQueryInfo) {
diff --git a/src/main/java/com/getindata/connectors/http/internal/table/lookup/HttpLookupConnectorOptions.java b/src/main/java/com/getindata/connectors/http/internal/table/lookup/HttpLookupConnectorOptions.java
@@ -39,6 +39,17 @@ public class HttpLookupConnectorOptions {
             .stringType()
             .noDefaultValue();
 
+    public static final ConfigOption<String> LOOKUP_HTTP_VERSION =
+        ConfigOptions.key(SOURCE_LOOKUP_QUERY_HTTP_VERSION)
+            .stringType()
+            .noDefaultValue()
+            .withDescription("Version of HTTP to use for lookup HTTP requests. " +
+                "The valid values are HTTP_1_1 and HTTP_2, which specify HTTP 1.1 or 2" +
+                " respectively. This option may be required as HTTP_1_1, if the" +
+                " endpoint is http 1.1, because some http 1.1 endpoints reject HTTP" +
+                " Version 2 calls, with 'Invalid HTTP request received' and " +
+                " 'HTTP/2 upgrade not supported'.");
+
     public static final ConfigOption<String> LOOKUP_REQUEST_FORMAT =
         ConfigOptions.key("lookup-request.format")
             .stringType()
diff --git a/src/main/java/com/getindata/connectors/http/internal/table/lookup/JavaNetHttpPollingClient.java b/src/main/java/com/getindata/connectors/http/internal/table/lookup/JavaNetHttpPollingClient.java
@@ -210,11 +210,9 @@ private HttpRowDataWrapper processHttpResponse(
             boolean isError) throws IOException {
 
         this.httpPostRequestCallback.call(response, request, "endpoint", Collections.emptyMap());
-
         var responseBody = response.body();
 
-        log.debug("Received status code [{}] for RestTableSource request with Server response body [{}] ",
-                response.statusCode(), responseBody);
+        log.debug("Received status code [{}] for RestTableSource request", response.statusCode());
         if (!isError && (StringUtils.isNullOrWhitespaceOnly(responseBody) || ignoreResponse(response))) {
             return HttpRowDataWrapper.builder()
                     .data(Collections.emptyList())
diff --git a/src/main/java/com/getindata/connectors/http/internal/table/lookup/RequestFactoryBase.java b/src/main/java/com/getindata/connectors/http/internal/table/lookup/RequestFactoryBase.java
@@ -1,10 +1,11 @@
 package com.getindata.connectors.http.internal.table.lookup;
 
+import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
 import java.net.http.HttpRequest.Builder;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.flink.annotation.VisibleForTesting;
@@ -39,6 +40,7 @@ public abstract class RequestFactoryBase implements HttpRequestFactory {
      */
     private final String[] headersAndValues;
     private final HttpLookupConfig options;
+    final HttpClient.Version httpVersion;
 
     public RequestFactoryBase(
             LookupQueryCreator lookupQueryCreator,
@@ -60,17 +62,18 @@ public RequestFactoryBase(
             );
 
         this.headersAndValues = HttpHeaderUtils.toHeaderAndValueArray(headerMap);
-
-        log.debug("RequestFactoryBase headersAndValues: " +
-                Arrays.stream(headersAndValues)
-                        .map(Object::toString)
-                        .collect(Collectors.joining(",")));
         this.httpRequestTimeOutSeconds = Integer.parseInt(
             options.getProperties().getProperty(
                 HttpConnectorConfigConstants.LOOKUP_HTTP_TIMEOUT_SECONDS,
                 DEFAULT_REQUEST_TIMEOUT_SECONDS
             )
         );
+        String httpVersionFromConfig = options.getReadableConfig().get(HttpLookupConnectorOptions.LOOKUP_HTTP_VERSION);
+        if (httpVersionFromConfig == null) {
+            httpVersion = null;
+        } else {
+            httpVersion = HttpClient.Version.valueOf(httpVersionFromConfig);
+        }
     }
 
     @Override
@@ -94,7 +97,14 @@ public HttpLookupSourceRequestEntry buildLookupRequest(RowData lookupRow) {
      * @param lookupQuery lookup query used for request query parameters or body.
      * @return {@link HttpRequest.Builder} for given lookupQuery.
      */
-    protected abstract Builder setUpRequestMethod(LookupQueryInfo lookupQuery);
+    protected Builder setUpRequestMethod(LookupQueryInfo lookupQuery) {
+        HttpRequest.Builder builder = HttpRequest.newBuilder()
+            .timeout(Duration.ofSeconds(this.httpRequestTimeOutSeconds));
+        if (httpVersion !=null) {
+            builder.version(httpVersion);
+        }
+        return builder;
+    }
 
     protected static StringBuilder resolvePathParameters(LookupQueryInfo lookupQueryInfo,
                                                          StringBuilder resolvedUrl) {
diff --git a/src/test/java/com/getindata/connectors/http/internal/table/lookup/BodyBasedRequestFactoryTest.java b/src/test/java/com/getindata/connectors/http/internal/table/lookup/BodyBasedRequestFactoryTest.java
@@ -1,33 +1,56 @@
 package com.getindata.connectors.http.internal.table.lookup;
 
+
 import java.net.URI;
+import java.net.http.HttpClient;
 import java.util.Collection;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 
+import org.apache.flink.configuration.Configuration;
 import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.testcontainers.shaded.com.google.common.collect.ImmutableList;
 import static org.assertj.core.api.Assertions.assertThat;
 
+import static com.getindata.connectors.http.internal.table.lookup.HttpLookupConnectorOptions.LOOKUP_HTTP_VERSION;
+
 public class BodyBasedRequestFactoryTest {
 
     @ParameterizedTest
     @MethodSource("configProvider")
     void testconstructUri(TestSpec testSpec) throws Exception {
-        LookupQueryInfo lookupQueryInfo = new LookupQueryInfo(testSpec.url,
+        Set<Configuration> configs = new HashSet();
+
+        Configuration configuration= new Configuration();
+        Configuration configuration_http11 = new Configuration();
+        Configuration configuration_http2 = new Configuration();
+
+        configuration_http2.setString(LOOKUP_HTTP_VERSION, String.valueOf(HttpClient.Version.HTTP_2));
+        configuration_http11.setString(LOOKUP_HTTP_VERSION, String.valueOf(HttpClient.Version.HTTP_1_1));
+
+        configs.add(configuration);
+        configs.add(configuration_http11);
+        configs.add(configuration_http2);
+
+        for(Configuration config: configs) {
+            LookupQueryInfo lookupQueryInfo = new LookupQueryInfo(testSpec.url,
                 testSpec.bodyBasedUrlQueryParams,
                 testSpec.pathBasedUrlParams);
-        HttpLookupConfig httpLookupConfig = HttpLookupConfig.builder()
+            HttpLookupConfig httpLookupConfig = HttpLookupConfig.builder()
                 .lookupMethod(testSpec.lookupMethod)
                 .url(testSpec.url)
                 .useAsync(false)
+                .readableConfig(config)
                 .build();
-        BodyBasedRequestFactory bodyBasedRequestFactory =
+            BodyBasedRequestFactory bodyBasedRequestFactory =
                 new BodyBasedRequestFactory("test", null, null, httpLookupConfig);
 
-        URI uri = bodyBasedRequestFactory.constructUri(lookupQueryInfo);
-        assertThat(uri.toString()).isEqualTo(testSpec.expected);
+            URI uri = bodyBasedRequestFactory.constructUri(lookupQueryInfo);
+            assertThat(uri.toString()).isEqualTo(testSpec.expected);
+        }
     }
 
     private static class TestSpec {
diff --git a/src/test/resources/simpleLogger.properties b/src/test/resources/simpleLogger.properties
@@ -0,0 +1,2 @@
+org.slf4j.simpleLogger.defaultLogLevel=INFO
+org.slf4j.simpleLogger.log.com.getindata.connectors.http.internal.table.lookup.RequestAndResponseLogger=DEBUG
