add max retries

Author: Raz0r

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

@@ -92,6 +92,12 @@ public final class HttpConnectorConfigConstants {
     public static final String SINK_HTTP_WRITER_THREAD_POOL_SIZE =
         GID_CONNECTOR_HTTP + "sink.writer.thread-pool.size";
 
+    public static final String LOOKUP_HTTP_MAX_RETRIES =
+        GID_CONNECTOR_HTTP + "source.lookup.request.max-retries";
+
+    public static final String LOOKUP_HTTP_RETRY_TIMEOUT_MS =
+        GID_CONNECTOR_HTTP + "source.lookup.request.retry-timeout-ms";
+
     // -----------------------------------------------------
 
 

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

@@ -27,6 +27,10 @@
 @Slf4j
 public class JavaNetHttpPollingClient implements PollingClient<RowData> {
 
+    public static final String DEFAULT_REQUEST_MAX_RETRIES = "3";
+
+    public static final String DEFAULT_REQUEST_RETRY_TIMEOUT_MS = "1000";
+
     private final HttpClient httpClient;
 
     private final HttpStatusCodeChecker statusCodeChecker;
@@ -37,6 +41,10 @@ public class JavaNetHttpPollingClient implements PollingClient<RowData> {
 
     private final HttpPostRequestCallback<HttpLookupSourceRequestEntry> httpPostRequestCallback;
 
+    protected final int httpRequestMaxRetries;
+
+    protected final int httpRequestRetryTimeoutMs;
+
     public JavaNetHttpPollingClient(
             HttpClient httpClient,
             DeserializationSchema<RowData> responseBodyDecoder,
@@ -62,6 +70,20 @@ public JavaNetHttpPollingClient(
                 .build();
 
         this.statusCodeChecker = new ComposeHttpStatusCodeChecker(checkerConfig);
+
+        this.httpRequestMaxRetries = Integer.parseInt(
+            options.getProperties().getProperty(
+                HttpConnectorConfigConstants.LOOKUP_HTTP_MAX_RETRIES,
+                DEFAULT_REQUEST_MAX_RETRIES
+            )
+        );
+
+        this.httpRequestRetryTimeoutMs = Integer.parseInt(
+            options.getProperties().getProperty(
+                HttpConnectorConfigConstants.LOOKUP_HTTP_RETRY_TIMEOUT_MS,
+                DEFAULT_REQUEST_RETRY_TIMEOUT_MS
+            )
+        );
     }
 
     @Override
@@ -74,15 +96,43 @@ public Optional<RowData> pull(RowData lookupRow) {
         }
     }
 
-    // TODO Add Retry Policy And configure TimeOut from properties
-    private Optional<RowData> queryAndProcess(RowData lookupData) throws Exception {
-
+    private Optional<RowData> queryAndProcess(RowData lookupData) {
         HttpLookupSourceRequestEntry request = requestFactory.buildLookupRequest(lookupData);
-        HttpResponse<String> response = httpClient.send(
-            request.getHttpRequest(),
-            BodyHandlers.ofString()
-        );
-        return processHttpResponse(response, request);
+        HttpResponse<String> response = null;
+
+        int retryCount = 0;
+
+        while (retryCount < this.httpRequestMaxRetries) {
+            try {
+                response = httpClient.send(
+                    request.getHttpRequest(),
+                    BodyHandlers.ofString()
+                );
+                break;
+            } catch (IOException e) {
+                log.error("IOException during HTTP request. Retrying...", e);
+                retryCount++;
+                if (retryCount == this.httpRequestMaxRetries) {
+                    log.error("Maximum retries reached. Aborting...");
+                    return Optional.empty();
+                }
+                try {
+                    Thread.sleep(this.httpRequestRetryTimeoutMs);
+                } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                log.error("HTTP request interrupted. Aborting...", e);
+                return Optional.empty();
+            }
+        }
+        try {
+            return processHttpResponse(response, request);
+        } catch (IOException e) {
+            log.error("IOException during HTTP response processing.", e);
+            return Optional.empty();
+        }
     }
 
     private Optional<RowData> processHttpResponse(

src/test/java/com/getindata/connectors/http/internal/table/lookup/JavaNetHttpPollingClientTest.java

@@ -1,5 +1,6 @@
 package com.getindata.connectors.http.internal.table.lookup;
 
+//import java.io.IOException;
 import java.net.URI;
 import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
@@ -22,7 +23,6 @@
 import org.mockito.junit.jupiter.MockitoExtension;
 import static org.assertj.core.api.Assertions.assertThat;
 
-
 import com.getindata.connectors.http.internal.HeaderPreprocessor;
 import com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants;
 import com.getindata.connectors.http.internal.table.lookup.querycreators.GenericGetQueryCreator;