Enable HttpPostRequestCallback to fail requests

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+### Added
+
+-   Added support for callbacks to mark requests as failed.
+
 ## [0.15.0] - 2024-07-30
 
 ### Added

README.md

@@ -375,6 +375,12 @@ and then reference identifier `rest-lookup-logger` in the HTTP lookup DDL proper
 is provided.
 
 
+- Callback Errors:
+
+   It is also possible to declare if a request should be considered failed from the [HttpPostRequestCallback](src/main/java/com/getindata/connectors/http/HttpPostRequestCallback.java) by throwing a 
+[PostRequestCallbackException](src/main/java/com/getindata/connectors/http/PostRequestCallbackException.java).
+
+
 ## 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.

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

@@ -25,5 +25,5 @@ void call(
         RequestT requestEntry,
         String endpointUrl,
         Map<String, String> headerMap
-    );
+    ) throws PostRequestCallbackException;
 }

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

@@ -0,0 +1,11 @@
+package com.getindata.connectors.http;
+
+public class PostRequestCallbackException extends Exception {
+    public PostRequestCallbackException(String message) {
+        super(message);
+    }
+
+    public PostRequestCallbackException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

src/main/java/com/getindata/connectors/http/internal/sink/httpclient/JavaNetSinkHttpClient.java

@@ -13,6 +13,7 @@
 import org.apache.flink.annotation.VisibleForTesting;
 
 import com.getindata.connectors.http.HttpPostRequestCallback;
+import com.getindata.connectors.http.PostRequestCallbackException;
 import com.getindata.connectors.http.internal.HeaderPreprocessor;
 import com.getindata.connectors.http.internal.SinkHttpClient;
 import com.getindata.connectors.http.internal.SinkHttpClientResponse;
@@ -98,13 +99,20 @@ private SinkHttpClientResponse prepareSinkHttpClientResponse(
         for (var response : responses) {
             var sinkRequestEntry = response.getHttpRequest();
             var optResponse = response.getResponse();
-
-            httpPostRequestCallback.call(
-                optResponse.orElse(null), sinkRequestEntry, endpointUrl, headerMap);
+            var failedCallback = false;
+
+            try {
+                httpPostRequestCallback.call(
+                        optResponse.orElse(null), sinkRequestEntry, endpointUrl, headerMap);
+            } catch (PostRequestCallbackException e) {
+                failedCallback = true;
+                log.debug("request marked as failed due to callback exception", e);
+            }
 
             // TODO Add response processor here and orchestrate it with statusCodeChecker.
             if (optResponse.isEmpty() ||
-                statusCodeChecker.isErrorCode(optResponse.get().statusCode())) {
+                statusCodeChecker.isErrorCode(optResponse.get().statusCode()) ||
+                failedCallback) {
                 failedResponses.add(sinkRequestEntry);
             } else {
                 successfulResponses.add(sinkRequestEntry);

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

@@ -14,6 +14,7 @@
 import org.apache.flink.util.StringUtils;
 
 import com.getindata.connectors.http.HttpPostRequestCallback;
+import com.getindata.connectors.http.PostRequestCallbackException;
 import com.getindata.connectors.http.internal.PollingClient;
 import com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants;
 import com.getindata.connectors.http.internal.status.ComposeHttpStatusCodeChecker;
@@ -89,7 +90,14 @@ private Optional<RowData> processHttpResponse(
             HttpResponse<String> response,
             HttpLookupSourceRequestEntry request) throws IOException {
 
-        this.httpPostRequestCallback.call(response, request, "endpoint", Collections.emptyMap());
+        try {
+            this.httpPostRequestCallback.call(
+                    response, request, "endpoint", Collections.emptyMap()
+            );
+        } catch (PostRequestCallbackException e) {
+            log.debug("Error during post request callback.", e);
+            return Optional.empty();
+        }
 
         if (response == null) {
             return Optional.empty();

src/test/java/com/getindata/connectors/http/internal/sink/httpclient/JavaNetSinkHttpClientConnectionTest.java

@@ -1,7 +1,9 @@
 package com.getindata.connectors.http.internal.sink.httpclient;
 
 import java.io.File;
+import java.net.http.HttpResponse;
 import java.util.Collections;
+import java.util.Map;
 import java.util.Properties;
 
 import com.github.tomakehurst.wiremock.WireMockServer;
@@ -19,6 +21,8 @@
 import static org.junit.jupiter.api.Assertions.assertAll;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
+import com.getindata.connectors.http.HttpPostRequestCallback;
+import com.getindata.connectors.http.PostRequestCallbackException;
 import com.getindata.connectors.http.internal.HttpsConnectionTestBase;
 import com.getindata.connectors.http.internal.SinkHttpClientResponse;
 import com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants;
@@ -62,6 +66,33 @@ public void testHttpConnection() {
             batchRequestSubmitterFactory);
     }
 
+    @Test
+    public void testHttpPostRequestCallbackWithException() {
+        wireMockServer = new WireMockServer(SERVER_PORT);
+        wireMockServer.start();
+        mockEndPoint(wireMockServer);
+
+        try {
+            JavaNetSinkHttpClient client =
+                    new JavaNetSinkHttpClient(
+                            properties,
+                            new TestPostRequestCallbackWithException(),
+                            headerPreprocessor,
+                            perRequestSubmitterFactory);
+            HttpSinkRequestEntry requestEntry = new HttpSinkRequestEntry("GET", new byte[0]);
+            SinkHttpClientResponse response =
+                    client.putRequests(
+                            Collections.singletonList(requestEntry),
+                            "https://localhost:" + HTTPS_SERVER_PORT + ENDPOINT
+                    ).get();
+
+            assertThat(response.getSuccessfulRequests()).isEmpty();
+            assertThat(response.getFailedRequests()).isNotEmpty();
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
     @Test
     public void testHttpsConnectionWithSelfSignedCert() {
 
@@ -366,4 +397,17 @@ private void mockEndPointWithBasicAuth(WireMockServer wireMockServer) {
                     .withBody("{}"))
         );
     }
+
+    public static class TestPostRequestCallbackWithException
+            implements HttpPostRequestCallback<HttpRequest> {
+        @Override
+        public void call(
+                HttpResponse<String> response,
+                HttpRequest requestEntry,
+                String endpointUrl,
+                Map<String, String> headerMap
+        ) throws PostRequestCallbackException {
+            throw new PostRequestCallbackException("Test exception");
+        }
+    }
 }