Introduce custom HTTP Sink Post Request Callback and its Factory (#23)

* Introduce custom HTTP Sink Post Request Callback

* Address review comments: remove 'Sink' from Callback class name

* Address review comments: make Request type generic

* Address review comments: accept null responses in callback

* Address review comments: set default callback in HttpSink

Author: swtwsk

CHANGELOG.md

@@ -2,15 +2,17 @@
 
 ## [Unreleased]
 
-### Fixed
-- Fix JavaDoc errors.
-
 ### Added
 - Add new properties `gid.connector.http.sink.error.code`,`gid.connector.http.sink.error.code.exclude`,
   `gid.connector.http.source.lookup.error.code` and `gid.connector.http.source.lookup.error.code.exclude`
   to set HTTP status codes that should be interpreted as errors both for HTTP Sink and HTTP Lookup Source.
-- Use Flink's format support to Http Lookup Source. 
+- Use Flink's format support to Http Lookup Source.
 - Add HTTP Lookup source client header configuration via properties.
+- Add [HttpPostRequestCallback](src/main/java/com/getindata/connectors/http/HttpPostRequestCallback.java) and
+  [HttpPostRequestCallbackFactory](src/main/java/com/getindata/connectors/http/HttpPostRequestCallbackFactory.java)
+  interfaces (along with a "default"
+  [Slf4jHttpPostRequestCallback](src/main/java/com/getindata/connectors/http/internal/table/sink/Slf4jHttpPostRequestCallback.java)
+  implementation) for customizable processing of HTTP Sink requests and responses in Table API.
 
 ### Changed
 - Change dependency scope for `org.apache.flink.flink-connector-base` from `compile` to `provided`.
@@ -20,6 +22,10 @@
 - Remove dependency on `org.apache.httpcomponents.httpclient`from production code. Dependency is only for test scope.
 - Removed dependency on `com.jayway.jsonpath.json-path`
 
+### Fixed
+
+- Fix JavaDoc errors.
+
 ## [0.3.0] - 2022-07-21
 
 -   Package refactoring. Hide internal classes that does not have to be used by API users under "internal" package.

README.md

@@ -184,6 +184,19 @@ CREATE TABLE http (
 )
 ```
 
+#### Custom request/response callback
+Http Sink 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 API Sink by implementing
+[HttpPostRequestCallback](src/main/java/com/getindata/connectors/http/HttpPostRequestCallback.java) and
+[HttpPostRequestCallbackFactory](src/main/java/com/getindata/connectors/http/HttpPostRequestCallbackFactory.java)
+interfaces. Custom implementations of `HttpSinkRequestCallbackFactory` can be registered along other factories in
+`resources/META-INF.services/org.apache.flink.table.factories.Factory` file and then referenced by their identifiers 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.
+
 ## HTTP status code handler
 Http Sink and Lookup Source connectors allow defining list of HTTP status codes that should be treated as errors. 
 By default all 400s and 500s response codes will be interpreted as error code.
@@ -201,25 +214,30 @@ An example of such a mask would be `3XX, 4XX, 5XX`. In this case, all 300s, 400s
 
 ## Table API Connector Options
 ### HTTP TableLookup Source
-| Option       | Required | Description/Value                                                                                        |
-|--------------|----------|----------------------------------------------------------------------------------------------------------|
-| connector    | required | The Value should be set to _rest-lookup_                                                                 |
-| 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.   |
+| Option                                       | Required | Description/Value                                                                                                             |
+|----------------------------------------------|----------|-------------------------------------------------------------------------------------------------------------------------------|
+| connector                                    | required | The Value should be set to _rest-lookup_                                                                                      |
+| 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.                        |
+| 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. |
 
 ### HTTP Sink
-| Option                     | Required | Description/Value                                                                                                                                                                                  |
-|----------------------------|----------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| connector                  | required | Specify what connector to use. For HTTP Sink it should be set to _'http-sink'_.                                                                                                                    |
-| url                        | required | The base URL that should be use for HTTP requests. For example _http://localhost:8080/client_.                                                                                                     |
-| format                     | required | Specify what format to use.                                                                                                                                                                        |
-| insert-method              | optional | Specify which HTTP method to use in the request. The value should be set either to `POST` or `PUT`.                                                                                                |
-| sink.batch.max-size        | optional | Maximum number of elements that may be passed in a batch to be written downstream.                                                                                                                 |
-| sink.requests.max-inflight | optional | The maximum number of in flight requests that may exist, if any more in flight requests need to be initiated once the maximum has been reached, then it will be blocked until some have completed. |
-| sink.requests.max-buffered | optional | Maximum number of buffered records before applying backpressure.                                                                                                                                   |
-| sink.flush-buffer.size     | optional | The maximum size of a batch of entries that may be sent to the HTTP endpoint measured in bytes.                                                                                                    |
-| sink.flush-buffer.timeout  | optional | Threshold time in milliseconds for an element to be in a buffer before being flushed.                                                                                                              |
+| Option                                     | Required | Description/Value                                                                                                                                                                                  |
+|--------------------------------------------|----------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| connector                                  | required | Specify what connector to use. For HTTP Sink it should be set to _'http-sink'_.                                                                                                                    |
+| url                                        | required | The base URL that should be use for HTTP requests. For example _http://localhost:8080/client_.                                                                                                     |
+| format                                     | required | Specify what format to use.                                                                                                                                                                        |
+| insert-method                              | optional | Specify which HTTP method to use in the request. The value should be set either to `POST` or `PUT`.                                                                                                |
+| sink.batch.max-size                        | optional | Maximum number of elements that may be passed in a batch to be written downstream.                                                                                                                 |
+| sink.requests.max-inflight                 | optional | The maximum number of in flight requests that may exist, if any more in flight requests need to be initiated once the maximum has been reached, then it will be blocked until some have completed. |
+| sink.requests.max-buffered                 | optional | Maximum number of buffered records before applying backpressure.                                                                                                                                   |
+| sink.flush-buffer.size                     | optional | The maximum size of a batch of entries that may be sent to the HTTP endpoint measured in bytes.                                                                                                    |
+| sink.flush-buffer.timeout                  | optional | Threshold time in milliseconds for an element to be in a buffer before being flushed.                                                                                                              |
+| gid.connector.http.sink.request-callback   | optional | Specify which `HttpPostRequestCallback` implementation to use. By default, it is set to `slf4j-logger` corresponding to `Slf4jHttpPostRequestCallback`.                                            |
+| gid.connector.http.sink.error.code         | optional | List of HTTP status codes that should be treated as errors by HTTP Sink, separated with comma.                                                                                                     |
+| gid.connector.http.sink.error.code.exclude | optional | List of HTTP status codes that should be excluded from the `gid.connector.http.sink.error.code` list, separated with comma.                                                                        |
 
 ## Build and deployment
 To build the project locally you need to have `maven 3` and Java 11+. </br>

pom.xml

@@ -325,6 +325,7 @@ under the License.
             <exclude>**/HttpLookupConnectorOptionsUtil.class</exclude>
             <exclude>**/HttpTableSourceFactoryHelper.class</exclude>
             <exclude>**/HttpPollTableSource.class</exclude>
+            <exclude>**/Slf4jHttpPostRequestCallback.class</exclude>
           </excludes>
         </configuration>
         <executions>

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

@@ -0,0 +1,29 @@
+package com.getindata.connectors.http;
+
+import java.io.Serializable;
+import java.net.http.HttpResponse;
+import java.util.Map;
+
+/**
+ * An interface for post request callback action, processing a response and its respective request.
+ *
+ * <p>One can customize the behaviour of such a callback by implementing both
+ * {@link HttpPostRequestCallback} and {@link HttpPostRequestCallbackFactory}.
+ *
+ * @param <RequestT> type of the HTTP request wrapper
+ */
+public interface HttpPostRequestCallback<RequestT> extends Serializable {
+    /**
+     * Process HTTP request and the matching response.
+     *  @param response HTTP response
+     * @param requestEntry request's payload
+     * @param endpointUrl the URL of the endpoint
+     * @param headerMap mapping of header names to header values
+     */
+    void call(
+        HttpResponse<String> response,
+        RequestT requestEntry,
+        String endpointUrl,
+        Map<String, String> headerMap
+    );
+}

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

@@ -0,0 +1,40 @@
+package com.getindata.connectors.http;
+
+import org.apache.flink.table.factories.Factory;
+
+import com.getindata.connectors.http.internal.table.sink.HttpDynamicSink;
+
+/**
+ * The {@link Factory} that dynamically creates and injects {@link HttpPostRequestCallback} to
+ * {@link HttpDynamicSink}.
+ *
+ * <p>Custom implementations of {@link HttpPostRequestCallbackFactory} can be registered along
+ * other factories in
+ * <pre>resources/META-INF.services/org.apache.flink.table.factories.Factory</pre>
+ * file and then referenced by their identifiers in the HttpSink DDL property field
+ * <i>gid.connector.http.sink.request-callback</i>.
+ *
+ * <p>The following example shows the minimum Table API example to create a {@link HttpDynamicSink}
+ * that uses a custom callback created by a factory that returns <i>my-callback</i> as its
+ * identifier.
+ *
+ * <pre>{@code
+ * CREATE TABLE http (
+ *   id bigint,
+ *   some_field string
+ * ) with (
+ *   'connector' = 'http-sink'
+ *   'url' = 'http://example.com/myendpoint'
+ *   'format' = 'json',
+ *   'gid.connector.http.sink.request-callback' = 'my-callback'
+ * )
+ * }</pre>
+ *
+ * @param <RequestT> type of the HTTP request wrapper
+ */
+public interface HttpPostRequestCallbackFactory<RequestT> extends Factory {
+    /**
+     * @return {@link HttpPostRequestCallback} custom request callback instance
+     */
+    HttpPostRequestCallback<RequestT> createHttpPostRequestCallback();
+}

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

@@ -40,6 +40,7 @@ public class HttpSink<InputT> extends HttpSinkInternal<InputT> {
             long maxTimeInBufferMS,
             long maxRecordSizeInBytes,
             String endpointUrl,
+            HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback,
             SinkHttpClientBuilder sinkHttpClientBuilder,
             Properties properties) {
         super(elementConverter,
@@ -50,6 +51,7 @@ public class HttpSink<InputT> extends HttpSinkInternal<InputT> {
             maxTimeInBufferMS,
             maxRecordSizeInBytes,
             endpointUrl,
+            httpPostRequestCallback,
             sinkHttpClientBuilder,
             properties
         );

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

@@ -11,6 +11,7 @@
 import com.getindata.connectors.http.internal.SinkHttpClientBuilder;
 import com.getindata.connectors.http.internal.sink.HttpSinkRequestEntry;
 import com.getindata.connectors.http.internal.sink.httpclient.JavaNetSinkHttpClient;
+import com.getindata.connectors.http.internal.table.sink.Slf4jHttpPostRequestCallback;
 
 /**
  * Builder to construct {@link HttpSink}.
@@ -59,6 +60,9 @@ public class HttpSinkBuilder<InputT> extends
 
     private static final SinkHttpClientBuilder DEFAULT_CLIENT_BUILDER = JavaNetSinkHttpClient::new;
 
+    private static final HttpPostRequestCallback<HttpSinkRequestEntry>
+        DEFAULT_POST_REQUEST_CALLBACK = new Slf4jHttpPostRequestCallback();
+
     private final Properties properties = new Properties();
 
     // Mandatory field
@@ -70,8 +74,12 @@ public class HttpSinkBuilder<InputT> extends
     // If not defined, should be set to DEFAULT_CLIENT_BUILDER
     private SinkHttpClientBuilder sinkHttpClientBuilder;
 
+    // If not defined, should be set to DEFAULT_POST_REQUEST_CALLBACK
+    private HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback;
+
     HttpSinkBuilder() {
         this.sinkHttpClientBuilder = DEFAULT_CLIENT_BUILDER;
+        this.httpPostRequestCallback = DEFAULT_POST_REQUEST_CALLBACK;
     }
 
     /**
@@ -105,6 +113,12 @@ public HttpSinkBuilder<InputT> setElementConverter(
         return this;
     }
 
+    public HttpSinkBuilder<InputT> setHttpPostRequestCallback(
+        HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback) {
+        this.httpPostRequestCallback = httpPostRequestCallback;
+        return this;
+    }
+
     /**
      * Set property for Http Sink.
      * @param propertyName property name
@@ -137,6 +151,7 @@ public HttpSink<InputT> build() {
             Optional.ofNullable(getMaxTimeInBufferMS()).orElse(DEFAULT_MAX_TIME_IN_BUFFER_MS),
             Optional.ofNullable(getMaxRecordSizeInBytes()).orElse(DEFAULT_MAX_RECORD_SIZE_IN_B),
             endpointUrl,
+            httpPostRequestCallback,
             sinkHttpClientBuilder,
             properties
         );

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

@@ -5,11 +5,16 @@
 
 import org.apache.flink.annotation.PublicEvolving;
 
+import com.getindata.connectors.http.HttpPostRequestCallback;
+import com.getindata.connectors.http.internal.sink.HttpSinkRequestEntry;
+
 /**
  * Builder building {@link SinkHttpClient}.
  */
 @PublicEvolving
 public interface SinkHttpClientBuilder extends Serializable {
-
-    SinkHttpClient build(Properties properties);
+    SinkHttpClient build(
+        Properties properties,
+        HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback
+    );
 }

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

@@ -37,7 +37,8 @@ public final class HttpConnectorConfigConstants {
 
     public static final String HTTP_ERROR_SOURCE_LOOKUP_CODES_LIST =
         GID_CONNECTOR_HTTP + "source.lookup.error.code";
-
     // -----------------------------------------------------
 
+    public static final String SINK_REQUEST_CALLBACK_IDENTIFIER =
+        GID_CONNECTOR_HTTP + "sink.request-callback";
 }