Merge pull request #46 from getindata/HTTP-45

[HTTP-45] - Fix issue with not printing HttpRequest body/parameters for Lookup Source by Slf4JHttpLookupPostRequestCallback.

Author: kristoffSC

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+### Fixed
+- Fixed issue with not printing HttpRequest body/parameters for Lookup Source by
+  [Slf4JHttpLookupPostRequestCallback](src/main/java/com/getindata/connectors/http/internal/table/lookup/Slf4JHttpLookupPostRequestCallback.java) - <https://github.com/getindata/flink-http-connector/issues/45>
+
 ### Removed
 -   Removed unused reference to EncodingFormat from HttpLookupTableSource
 

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

@@ -0,0 +1,33 @@
+package com.getindata.connectors.http.internal.table.lookup;
+
+import java.net.http.HttpRequest;
+
+import lombok.Data;
+import lombok.ToString;
+
+/**
+ * Wrapper class around {@link HttpRequest} that contains information about an actual lookup request
+ * body or request parameters.
+ */
+@Data
+@ToString
+public class HttpLookupSourceRequestEntry {
+
+    /**
+     * Wrapped {@link HttpRequest} object.
+     */
+    private final HttpRequest httpRequest;
+
+    /**
+     * This field represents lookup query. Depending on used REST request method, this field can
+     * represent a request body, for example a Json string when PUT/POST requests method was used,
+     * or it can represent a query parameters if GET method was used.
+     */
+    private final String lookupQuery;
+
+    public HttpLookupSourceRequestEntry(HttpRequest httpRequest, String lookupQuery) {
+
+        this.httpRequest = httpRequest;
+        this.lookupQuery = lookupQuery;
+    }
+}

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

@@ -16,5 +16,5 @@ public interface HttpRequestFactory extends Serializable {
      * @param lookupRow {@link RowData} object used for building http request.
      * @return {@link HttpRequest} created from {@link RowData}
      */
-    HttpRequest buildLookupRequest(RowData lookupRow);
+    HttpLookupSourceRequestEntry buildLookupRequest(RowData lookupRow);
 }

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

@@ -2,7 +2,6 @@
 
 import java.io.IOException;
 import java.net.http.HttpClient;
-import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
 import java.net.http.HttpResponse.BodyHandlers;
 import java.util.Collections;
@@ -36,7 +35,7 @@ public class JavaNetHttpPollingClient implements PollingClient<RowData> {
 
     private final HttpRequestFactory requestFactory;
 
-    private final HttpPostRequestCallback<HttpRequest> httpPostRequestCallback;
+    private final HttpPostRequestCallback<HttpLookupSourceRequestEntry> httpPostRequestCallback;
 
     public JavaNetHttpPollingClient(
             HttpClient httpClient,
@@ -78,19 +77,22 @@ public Optional<RowData> pull(RowData lookupRow) {
     // TODO Add Retry Policy And configure TimeOut from properties
     private Optional<RowData> queryAndProcess(RowData lookupData) throws Exception {
 
-        HttpRequest request = requestFactory.buildLookupRequest(lookupData);
-        HttpResponse<String> response = httpClient.send(request, BodyHandlers.ofString());
+        HttpLookupSourceRequestEntry request = requestFactory.buildLookupRequest(lookupData);
+        HttpResponse<String> response = httpClient.send(
+            request.getHttpRequest(),
+            BodyHandlers.ofString()
+        );
         return processHttpResponse(response, request);
     }
 
     private Optional<RowData> processHttpResponse(
             HttpResponse<String> response,
-            HttpRequest request) throws IOException {
+            HttpLookupSourceRequestEntry request) throws IOException {
 
         this.httpPostRequestCallback.call(response, request, "endpoint", Collections.emptyMap());
 
         if (response == null) {
-            log.warn("Null Http response for request " + request.uri().toString());
+            log.warn("Null Http response for request " + request.getHttpRequest().uri().toString());
             return Optional.empty();
         }
 

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

@@ -59,7 +59,7 @@ public RequestFactoryBase(
     }
 
     @Override
-    public HttpRequest buildLookupRequest(RowData lookupRow) {
+    public HttpLookupSourceRequestEntry buildLookupRequest(RowData lookupRow) {
 
         String lookupQuery = lookupQueryCreator.createLookupQuery(lookupRow);
         getLogger().debug("Created Http lookup query: " + lookupQuery);
@@ -70,7 +70,7 @@ public HttpRequest buildLookupRequest(RowData lookupRow) {
             requestBuilder.headers(headersAndValues);
         }
 
-        return requestBuilder.build();
+        return new HttpLookupSourceRequestEntry(requestBuilder.build(), lookupQuery);
     }
 
     protected abstract Logger getLogger();

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

@@ -20,17 +20,20 @@
  * the {@link HttpLookupTableSource}.
  */
 @Slf4j
-public class Slf4JHttpLookupPostRequestCallback implements HttpPostRequestCallback<HttpRequest> {
+public class Slf4JHttpLookupPostRequestCallback
+        implements HttpPostRequestCallback<HttpLookupSourceRequestEntry> {
 
     @Override
     public void call(
             HttpResponse<String> response,
-            HttpRequest requestEntry,
+            HttpLookupSourceRequestEntry requestEntry,
             String endpointUrl,
             Map<String, String> headerMap) {
 
+        HttpRequest httpRequest = requestEntry.getHttpRequest();
         StringJoiner headers = new StringJoiner(";");
-        for (Entry<String, List<String>> reqHeaders : requestEntry.headers().map().entrySet()) {
+
+        for (Entry<String, List<String>> reqHeaders : httpRequest.headers().map().entrySet()) {
             StringJoiner values = new StringJoiner(";");
             for (String value : reqHeaders.getValue()) {
                 values.add(value);
@@ -41,17 +44,21 @@ public void call(
 
         if (response == null) {
             log.info(
-                "Got response for a request.\n  Request:\n    " +
-                    "Method: {}\n    Headers: {}\n    Body: {}\n    Response: null",
-                requestEntry.method(), headers, requestEntry
+                "Got response for a request.\n  Request:\n    URL: {}\n    " +
+                    "Method: {}\n    Headers: {}\n    Params/Body: {}\nResponse: null",
+                httpRequest.uri().toString(),
+                httpRequest.method(),
+                headers,
+                requestEntry.getLookupQuery()
             );
         } else {
             log.info(
-                "Got response for a request.\n  Request:\n    " +
-                    "Method: {}\n    Headers: {}\n    Body: {}\n    Response: {}\n    Body: {}",
-                requestEntry.method(),
+                "Got response for a request.\n  Request:\n    URL: {}\n    " +
+                    "Method: {}\n    Headers: {}\n    Params/Body: {}\nResponse: {}\n    Body: {}",
+                httpRequest.uri().toString(),
+                httpRequest.method(),
                 headers,
-                requestEntry,
+                requestEntry.getLookupQuery(),
                 response,
                 response.body().replaceAll(ConfigUtils.UNIVERSAL_NEW_LINE_REGEXP, "")
             );