[HTTP-38] - Fixing issue with handling Http timeout where Event task was not removed from Flink's Operator AsyncIO internal queue.

Signed-off-by: Krzysztof Chmielewski <krzysiek.chmielewski@gmail.com>

Author: kristoffSC

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## [Unreleased]
 
+### Added
+-   Add new parameters for HTTP timeout configuration and thread pool size for Sink and Lookup source http requests.
+
+### Fixed
+-  Fix issue with not cleaning Flink's internal task queue for AsyncIO requests after HTTP timeout in
+   Lookup source - https://github.com/getindata/flink-http-connector/issues/38
+
 ## [0.7.0] - 2022-10-27
 
 -   Add to Lookup Source support for performing lookup on columns with complex types such as ROW, Map etc.

README.md

@@ -65,4 +65,23 @@ public final class HttpConnectorConfigConstants {
         + "security.keystore.type";
 
     // -----------------------------------------------------
+
+    // ------ HTTPS timeouts and thread pool settings ------
+
+    public static final String LOOKUP_HTTP_TIMEOUT_SECONDS =
+        GID_CONNECTOR_HTTP + "source.lookup.request.timeout";
+
+    public static final String SINK_HTTP_TIMEOUT_SECONDS =
+        GID_CONNECTOR_HTTP + "sink.request.timeout";
+
+    public static final String LOOKUP_HTTP_PULING_THREAD_POOL_SIZE =
+        GID_CONNECTOR_HTTP + "source.lookup.request.thread-pool.size";
+
+    public static final String LOOKUP_HTTP_RESPONSE_THREAD_POOL_SIZE =
+        GID_CONNECTOR_HTTP + "source.lookup.response.thread-pool.size";
+
+    public static final String SINK_HTTP_WRITER_THREAD_POOL_SIZE =
+        GID_CONNECTOR_HTTP + "sink.writer.thread-pool.size";
+
+    // -----------------------------------------------------
 }

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

@@ -24,6 +24,8 @@
 
 import lombok.extern.slf4j.Slf4j;
 
+import com.getindata.connectors.http.internal.utils.ConfigUtils;
+
 /**
  * This class represents a security context for given Http connector instance. The Security context
  * is backed by in memory instance of Java's {@link KeyStore}. All keys and certificates managed by
@@ -42,17 +44,6 @@ public class SecurityContext {
 
     private static final String PRIVATE_KEY_FOOTER = "-----END PRIVATE KEY-----";
 
-    /**
-     * A pattern matcher linebreak regexp that represents any Unicode linebreak sequence making it
-     * effectively equivalent to:
-     * <pre>
-     * {@code
-     * &#92;u000D&#92;u000A|[&#92;u000A&#92;u000B&#92;u000C&#92;u000D&#92;u0085&#92;u2028&#92;u2029]
-     * }
-     * </pre>
-     */
-    public static final String UNIVERSAL_NEW_LINE_REGEXP = "\\R";
-
     private final char[] storePassword;
 
     private final KeyStore keystore;
@@ -228,7 +219,7 @@ private byte[] decodePrivateData(String privateKeyPath, byte[] privateData) {
         if (privateKeyPath.endsWith(".pem")) {
             String privateString = new String(privateData, Charset.defaultCharset())
                 .replace(PRIVATE_KEY_HEADER, "")
-                .replaceAll(UNIVERSAL_NEW_LINE_REGEXP, "")
+                .replaceAll(ConfigUtils.UNIVERSAL_NEW_LINE_REGEXP, "")
                 .replace(PRIVATE_KEY_FOOTER, "");
 
             return Base64.getDecoder().decode(privateString);

src/main/java/com/getindata/connectors/http/internal/security/SecurityContext.java

@@ -108,8 +108,8 @@ protected HttpSinkInternal(
 
     @Override
     public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> createWriter(
-        InitContext context
-    ) throws IOException {
+            InitContext context) throws IOException {
+
         return new HttpSinkWriter<>(
             getElementConverter(),
             context,
@@ -121,14 +121,17 @@ public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> cr
             getMaxRecordSizeInBytes(),
             endpointUrl,
             sinkHttpClientBuilder.build(properties, httpPostRequestCallback, headerPreprocessor),
-            Collections.emptyList()
+            Collections.emptyList(),
+            properties
         );
     }
 
     @Override
     public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> restoreWriter(
-        InitContext context, Collection<BufferedRequestState<HttpSinkRequestEntry>> recoveredState
-    ) throws IOException {
+            InitContext context,
+            Collection<BufferedRequestState<HttpSinkRequestEntry>> recoveredState)
+            throws IOException {
+
         return new HttpSinkWriter<>(
             getElementConverter(),
             context,
@@ -140,7 +143,8 @@ public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> re
             getMaxRecordSizeInBytes(),
             endpointUrl,
             sinkHttpClientBuilder.build(properties, httpPostRequestCallback, headerPreprocessor),
-            recoveredState
+            recoveredState,
+            properties
         );
     }
 

src/main/java/com/getindata/connectors/http/internal/sink/HttpSinkInternal.java

@@ -3,6 +3,9 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.function.Consumer;
 
 import lombok.extern.slf4j.Slf4j;
@@ -11,8 +14,11 @@
 import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
 import org.apache.flink.connector.base.sink.writer.ElementConverter;
 import org.apache.flink.metrics.Counter;
+import org.apache.flink.util.concurrent.ExecutorThreadFactory;
 
 import com.getindata.connectors.http.internal.SinkHttpClient;
+import com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants;
+import com.getindata.connectors.http.internal.utils.ThreadUtils;
 
 /**
  * Sink writer created by {@link com.getindata.connectors.http.HttpSink} to write to an HTTP
@@ -26,6 +32,13 @@
 @Slf4j
 public class HttpSinkWriter<InputT> extends AsyncSinkWriter<InputT, HttpSinkRequestEntry> {
 
+    private static final String HTTP_SINK_WRITER_THREAD_POOL_SIZE = "4";
+
+    /**
+     * Thread pool to handle HTTP response from HTTP client.
+     */
+    private final ExecutorService sinkWriterThreadPool;
+
     private final String endpointUrl;
 
     private final SinkHttpClient sinkHttpClient;
@@ -43,14 +56,27 @@ public HttpSinkWriter(
             long maxRecordSizeInBytes,
             String endpointUrl,
             SinkHttpClient sinkHttpClient,
-            Collection<BufferedRequestState<HttpSinkRequestEntry>> bufferedRequestStates) {
+            Collection<BufferedRequestState<HttpSinkRequestEntry>> bufferedRequestStates,
+            Properties properties) {
+
         super(elementConverter, context, maxBatchSize, maxInFlightRequests, maxBufferedRequests,
             maxBatchSizeInBytes, maxTimeInBufferMS, maxRecordSizeInBytes, bufferedRequestStates);
         this.endpointUrl = endpointUrl;
         this.sinkHttpClient = sinkHttpClient;
 
         var metrics = context.metricGroup();
         this.numRecordsSendErrorsCounter = metrics.getNumRecordsSendErrorsCounter();
+
+        int sinkWriterThreadPollSize = Integer.parseInt(properties.getProperty(
+            HttpConnectorConfigConstants.SINK_HTTP_WRITER_THREAD_POOL_SIZE,
+            HTTP_SINK_WRITER_THREAD_POOL_SIZE
+        ));
+
+        this.sinkWriterThreadPool =
+            Executors.newFixedThreadPool(
+                sinkWriterThreadPollSize,
+                new ExecutorThreadFactory(
+                    "http-sink-writer-worker", ThreadUtils.LOGGING_EXCEPTION_HANDLER));
     }
 
     // TODO: Reintroduce retries by adding backoff policy
@@ -59,7 +85,7 @@ protected void submitRequestEntries(
             List<HttpSinkRequestEntry> requestEntries,
             Consumer<List<HttpSinkRequestEntry>> requestResult) {
         var future = sinkHttpClient.putRequests(requestEntries, endpointUrl);
-        future.whenComplete((response, err) -> {
+        future.whenCompleteAsync((response, err) -> {
             if (err != null) {
                 int failedRequestsNumber = requestEntries.size();
                 log.error(
@@ -89,11 +115,17 @@ protected void submitRequestEntries(
                 //}
             }
             requestResult.accept(Collections.emptyList());
-        });
+        }, sinkWriterThreadPool);
     }
 
     @Override
     protected long getSizeInBytes(HttpSinkRequestEntry s) {
         return s.getSizeInBytes();
     }
+
+    @Override
+    public void close() {
+        sinkWriterThreadPool.shutdownNow();
+        super.close();
+    }
 }

src/main/java/com/getindata/connectors/http/internal/sink/HttpSinkWriter.java

@@ -7,12 +7,20 @@
 import java.net.http.HttpRequest.BodyPublishers;
 import java.net.http.HttpRequest.Builder;
 import java.net.http.HttpResponse;
-import java.util.*;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.stream.Collectors;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.concurrent.ExecutorThreadFactory;
 
 import com.getindata.connectors.http.HttpPostRequestCallback;
 import com.getindata.connectors.http.internal.HeaderPreprocessor;
@@ -26,14 +34,19 @@
 import com.getindata.connectors.http.internal.table.sink.Slf4jHttpPostRequestCallback;
 import com.getindata.connectors.http.internal.utils.HttpHeaderUtils;
 import com.getindata.connectors.http.internal.utils.JavaNetHttpClientFactory;
+import com.getindata.connectors.http.internal.utils.ThreadUtils;
 
 /**
- * An implementation of {@link SinkHttpClient} that uses Java 11's {@link HttpClient}.
- * This implementation supports HTTP traffic only.
+ * An implementation of {@link SinkHttpClient} that uses Java 11's {@link HttpClient}. This
+ * implementation supports HTTP traffic only.
  */
 @Slf4j
 public class JavaNetSinkHttpClient implements SinkHttpClient {
 
+    private static final int HTTP_CLIENT_THREAD_POOL_SIZE = 16;
+
+    public static final String DEFAULT_REQUEST_TIMEOUT_SECONDS = "30";
+
     private final HttpClient httpClient;
 
     private final String[] headersAndValues;
@@ -44,16 +57,29 @@ public class JavaNetSinkHttpClient implements SinkHttpClient {
 
     private final HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback;
 
+    /**
+     * Thread pool to handle HTTP response from HTTP client.
+     */
+    private final ExecutorService publishingThreadPool;
+
+    private final int httpRequestTimeOutSeconds;
+
     public JavaNetSinkHttpClient(Properties properties, HeaderPreprocessor headerPreprocessor) {
         this(properties, new Slf4jHttpPostRequestCallback(), headerPreprocessor);
     }
 
     public JavaNetSinkHttpClient(
-            Properties properties,
-            HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback,
-            HeaderPreprocessor headerPreprocessor) {
+        Properties properties,
+        HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback,
+        HeaderPreprocessor headerPreprocessor) {
 
-        this.httpClient = JavaNetHttpClientFactory.createClient(properties);
+        ExecutorService httpClientExecutor =
+            Executors.newFixedThreadPool(
+                HTTP_CLIENT_THREAD_POOL_SIZE,
+                new ExecutorThreadFactory(
+                    "http-sink-client-request-worker", ThreadUtils.LOGGING_EXCEPTION_HANDLER));
+
+        this.httpClient = JavaNetHttpClientFactory.createClient(properties, httpClientExecutor);
         this.httpPostRequestCallback = httpPostRequestCallback;
         this.headerMap = HttpHeaderUtils.prepareHeaderMap(
             HttpConnectorConfigConstants.SINK_HEADER_PREFIX,
@@ -72,12 +98,23 @@ public JavaNetSinkHttpClient(
                 .build();
 
         this.statusCodeChecker = new ComposeHttpStatusCodeChecker(checkerConfig);
+
+        this.publishingThreadPool =
+            Executors.newFixedThreadPool(
+                HTTP_CLIENT_THREAD_POOL_SIZE,
+                new ExecutorThreadFactory(
+                    "http-sink-client-response-worker", ThreadUtils.LOGGING_EXCEPTION_HANDLER));
+
+        this.httpRequestTimeOutSeconds = Integer.parseInt(
+            properties.getProperty(HttpConnectorConfigConstants.SINK_HTTP_TIMEOUT_SECONDS,
+                DEFAULT_REQUEST_TIMEOUT_SECONDS)
+        );
     }
 
     @Override
     public CompletableFuture<SinkHttpClientResponse> putRequests(
-            List<HttpSinkRequestEntry> requestEntries,
-            String endpointUrl) {
+        List<HttpSinkRequestEntry> requestEntries,
+        String endpointUrl) {
         return submitRequests(requestEntries, endpointUrl)
             .thenApply(responses -> prepareSinkHttpClientResponse(responses, endpointUrl));
     }
@@ -87,6 +124,7 @@ private HttpRequest buildHttpRequest(HttpSinkRequestEntry requestEntry, URI endp
             .newBuilder()
             .uri(endpointUri)
             .version(Version.HTTP_1_1)
+            .timeout(Duration.ofSeconds(httpRequestTimeOutSeconds))
             .method(requestEntry.method,
                 BodyPublishers.ofByteArray(requestEntry.element));
 
@@ -98,20 +136,25 @@ private HttpRequest buildHttpRequest(HttpSinkRequestEntry requestEntry, URI endp
     }
 
     private CompletableFuture<List<JavaNetHttpResponseWrapper>> submitRequests(
-            List<HttpSinkRequestEntry> requestEntries,
-            String endpointUrl) {
+        List<HttpSinkRequestEntry> requestEntries,
+        String endpointUrl) {
         var endpointUri = URI.create(endpointUrl);
         var responseFutures = new ArrayList<CompletableFuture<JavaNetHttpResponseWrapper>>();
 
         for (var entry : requestEntries) {
             var response = httpClient
-                .sendAsync(buildHttpRequest(entry, endpointUri),
+                .sendAsync(
+                    buildHttpRequest(entry, endpointUri),
                     HttpResponse.BodyHandlers.ofString())
                 .exceptionally(ex -> {
+                    // TODO This will be executed on a ForJoinPool Thread... refactor this someday.
                     log.error("Request fatally failed because of an exception", ex);
                     return null;
                 })
-                .thenApply(res -> new JavaNetHttpResponseWrapper(entry, res));
+                .thenApplyAsync(
+                    res -> new JavaNetHttpResponseWrapper(entry, res),
+                    publishingThreadPool
+                );
             responseFutures.add(response);
         }
 
@@ -121,8 +164,8 @@ private CompletableFuture<List<JavaNetHttpResponseWrapper>> submitRequests(
     }
 
     private SinkHttpClientResponse prepareSinkHttpClientResponse(
-            List<JavaNetHttpResponseWrapper> responses,
-            String endpointUrl) {
+        List<JavaNetHttpResponseWrapper> responses,
+        String endpointUrl) {
         var successfulResponses = new ArrayList<HttpSinkRequestEntry>();
         var failedResponses = new ArrayList<HttpSinkRequestEntry>();