add test

Author: Raz0r

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

@@ -96,43 +96,14 @@ public Optional<RowData> pull(RowData lookupRow) {
         }
     }
 
-    private Optional<RowData> queryAndProcess(RowData lookupData) {
+    private Optional<RowData> queryAndProcess(RowData lookupData) throws Exception {
+
         HttpLookupSourceRequestEntry request = requestFactory.buildLookupRequest(lookupData);
-        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();
-        }
+        HttpResponse<String> response = httpClient.send(
+            request.getHttpRequest(),
+            BodyHandlers.ofString()
+        );
+        return processHttpResponse(response, request);
     }
 
     private Optional<RowData> processHttpResponse(

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

@@ -7,7 +7,9 @@
 
 import com.github.tomakehurst.wiremock.WireMockServer;
 import com.github.tomakehurst.wiremock.client.MappingBuilder;
+import com.github.tomakehurst.wiremock.http.Fault;
 import com.github.tomakehurst.wiremock.matching.RequestPatternBuilder;
+import com.github.tomakehurst.wiremock.stubbing.Scenario;
 import com.github.tomakehurst.wiremock.stubbing.StubMapping;
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 import org.apache.flink.api.common.serialization.SerializationSchema;
@@ -290,6 +292,45 @@ public void shouldConnectWithBasicAuth(String authorizationHeaderValue,
         assertThat(nestedDetailsRow.getString(0).toString()).isEqualTo("$1,729.34");
     }
 
+    @Test
+    void shouldRetryOnIOExceptionAndSucceedOnSecondAttempt() {
+        // GIVEN
+        this.stubMapping = setUpServerStubForIOExceptionOnFirstAttempt();
+        Properties properties = new Properties();
+        properties.setProperty(
+            HttpConnectorConfigConstants.LOOKUP_HTTP_MAX_RETRIES,
+            "3"
+        );
+        JavaNetHttpPollingClient pollingClient = setUpPollingClient(
+            getBaseUrl(), properties, setUpGetRequestFactory(properties));
+
+        // WHEN
+        Optional<RowData> poll = pollingClient.pull(lookupRowData);
+
+        // THEN
+        wireMockServer.verify(2, RequestPatternBuilder.forCustomMatcher(stubMapping.getRequest()));
+
+        assertThat(poll.isPresent()).isTrue();
+    }
+
+    private StubMapping setUpServerStubForIOExceptionOnFirstAttempt() {
+        wireMockServer.stubFor(
+            get(urlEqualTo(ENDPOINT + "?id=1&uuid=2"))
+                .inScenario("Retry Scenario")
+                .whenScenarioStateIs(Scenario.STARTED) // Initial state
+                .willReturn(aResponse()
+                    .withFault(Fault.CONNECTION_RESET_BY_PEER)) // Fail the first request
+                .willSetStateTo("Second Attempt")); // Set the next state
+
+        return wireMockServer.stubFor(
+            get(urlEqualTo(ENDPOINT + "?id=1&uuid=2"))
+                .inScenario("Retry Scenario")
+                .whenScenarioStateIs("Second Attempt") // When the state is "Second Attempt"
+                .willReturn(aResponse()
+                    .withStatus(200)
+                    .withBody(readTestFile(SAMPLES_FOLDER + "HttpResult.json"))));
+    }
+
     private String getBaseUrl() {
         return wireMockServer.baseUrl() + ENDPOINT;
     }