Merge branch 'main' into http179

Author: grzegorz8

CHANGELOG.md

@@ -2,7 +2,10 @@
 
 ## [Unreleased]
 
+
 - Ability to specify http versions for http lookups. 
+- Amend to not log HTTP request response and header values by default.
+- Added http 2 support.
 
 ## [0.22.0] - 2025-10-03
 

README.md

@@ -427,10 +427,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

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())

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

@@ -6,7 +6,6 @@
 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;
@@ -63,11 +62,6 @@ 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,

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

@@ -2,15 +2,11 @@
 
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
-import java.util.List;
 import java.util.Map;
-import java.util.Map.Entry;
-import java.util.StringJoiner;
 
 import lombok.extern.slf4j.Slf4j;
 
 import com.getindata.connectors.http.HttpPostRequestCallback;
-import com.getindata.connectors.http.internal.utils.ConfigUtils;
 
 /**
  * A {@link HttpPostRequestCallback} that logs pairs of request and response as <i>INFO</i> level
@@ -31,40 +27,26 @@ public void call(
             Map<String, String> headerMap) {
 
         HttpRequest httpRequest = requestEntry.getHttpRequest();
-        StringJoiner headers = new StringJoiner(";");
-
-        for (Entry<String, List<String>> reqHeaders : httpRequest.headers().map().entrySet()) {
-            StringJoiner values = new StringJoiner(";");
-            for (String value : reqHeaders.getValue()) {
-                values.add(value);
-            }
-            String header = reqHeaders.getKey() + ": [" + values + "]";
-            headers.add(header);
-        }
 
         if (response == null) {
             log.warn("Null Http response for request " + httpRequest.uri().toString());
 
             log.info(
                 "Got response for a request.\n  Request:\n    URL: {}\n    " +
-                    "Method: {}\n    Headers: {}\n    Params/Body: {}\nResponse: null",
+                    "Method: {}\n Params/Body: {}\nResponse: null",
                 httpRequest.uri().toString(),
                 httpRequest.method(),
-                headers,
                 requestEntry.getLookupQueryInfo()
             );
         } else {
             log.info(
                 "Got response for a request.\n  Request:\n    URL: {}\n    " +
-                    "Method: {}\n    Headers: {}\n    Params/Body: {}\nResponse: {}\n    Body: {}",
+                    "Method: {}\n Params/Body: {}\nResponse status code: {}\n",
                 httpRequest.uri().toString(),
                 httpRequest.method(),
-                headers,
                 requestEntry.getLookupQueryInfo(),
-                response,
-                response.body().replaceAll(ConfigUtils.UNIVERSAL_NEW_LINE_REGEXP, "")
+                response.statusCode()
             );
         }
-
     }
 }

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

@@ -1,19 +1,17 @@
 package com.getindata.connectors.http.internal.table.sink;
 
 import java.net.http.HttpResponse;
-import java.nio.charset.StandardCharsets;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 import lombok.extern.slf4j.Slf4j;
 
 import com.getindata.connectors.http.HttpPostRequestCallback;
 import com.getindata.connectors.http.internal.sink.httpclient.HttpRequest;
-import com.getindata.connectors.http.internal.utils.ConfigUtils;
 
 /**
  * A {@link HttpPostRequestCallback} that logs pairs of request and response as <i>INFO</i> level
- * logs using <i>Slf4j</i>.
+ * logs using <i>Slf4j</i>. As the request/response body or header might contain sensitive information,
+ * we do not log those values.
  *
  * <p>Serving as a default implementation of {@link HttpPostRequestCallback} for
  * the {@link HttpDynamicSink}.
@@ -28,25 +26,23 @@ public void call(
         String endpointUrl,
         Map<String, String> headerMap) {
 
-        String requestBody = requestEntry.getElements().stream()
-            .map(element -> new String(element, StandardCharsets.UTF_8))
-            .collect(Collectors.joining());
+        // Uncomment if you want to see the requestBody in the log
+        //String requestBody = requestEntry.getElements().stream()
+        //    .map(element -> new String(element, StandardCharsets.UTF_8))
+        //    .collect(Collectors.joining());
 
         if (response == null) {
             log.info(
                 "Got response for a request.\n  Request:\n    " +
-                "Method: {}\n    Body: {}\n  Response: null",
-                requestEntry.getMethod(),
-                requestBody
+                "Method: {}\n   Response: null",
+                requestEntry.getMethod()
             );
         } else {
             log.info(
                 "Got response for a request.\n  Request:\n    " +
-                "Method: {}\n    Body: {}\n  Response: {}\n    Body: {}",
+                "Method: {}\n  Response status code: {}\n ",
                 requestEntry.method,
-                requestBody,
-                response,
-                response.body().replaceAll(ConfigUtils.UNIVERSAL_NEW_LINE_REGEXP, "")
+                response.statusCode()
             );
         }
     }

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