Revert Sink changes

Author: Grzegorz Kołakowski

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

@@ -1,15 +1,13 @@
 package com.getindata.connectors.http.internal;
 
 import java.util.List;
-import java.util.stream.Collectors;
 
 import lombok.Data;
 import lombok.NonNull;
 import lombok.ToString;
 
 import com.getindata.connectors.http.internal.sink.HttpSinkRequestEntry;
 import com.getindata.connectors.http.internal.sink.httpclient.HttpRequest;
-import com.getindata.connectors.http.internal.status.HttpResponseStatus;
 
 /**
  * Data class holding {@link HttpSinkRequestEntry} instances that {@link SinkHttpClient} attempted
@@ -20,36 +18,14 @@
 public class SinkHttpClientResponse {
 
     /**
-     * A list of requests along with write status.
+     * A list of successfully written requests.
      */
     @NonNull
-    private final List<ResponseItem> requests;
+    private final List<HttpRequest> successfulRequests;
 
-    public List<HttpRequest> getSuccessfulRequests() {
-        return requests.stream()
-                .filter(i -> i.getStatus() == HttpResponseStatus.SUCCESS)
-                .map(ResponseItem::getRequest)
-                .collect(Collectors.toList());
-    }
-
-    public List<HttpRequest> getFailedRetryableRequests() {
-        return requests.stream()
-                .filter(i -> i.getStatus() == HttpResponseStatus.FAILURE_RETRYABLE)
-                .map(ResponseItem::getRequest)
-                .collect(Collectors.toList());
-    }
-
-    public List<HttpRequest> getFailedNotRetryableRequests() {
-        return requests.stream()
-                .filter(i -> i.getStatus() == HttpResponseStatus.FAILURE_NOT_RETRYABLE)
-                .map(ResponseItem::getRequest)
-                .collect(Collectors.toList());
-    }
-
-    @Data
-    @ToString
-    public static class ResponseItem {
-        private final HttpRequest request;
-        private final HttpResponseStatus status;
-    }
+    /**
+     * A list of requests that {@link SinkHttpClient} failed to write.
+     */
+    @NonNull
+    private final List<HttpRequest> failedRequests;
 }

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

@@ -98,10 +98,8 @@ protected void submitRequestEntries(
                 //  to the `numRecordsSendErrors` metric. It is due to the fact we do not have
                 //  a clear image how we want to do it, so it would be both efficient and correct.
                 //requestResult.accept(requestEntries);
-            } else if (response.getFailedNotRetryableRequests().size()
-                + response.getFailedRetryableRequests().size() > 0) {
-                int failedRequestsNumber = response.getFailedNotRetryableRequests().size()
-                    + response.getFailedRetryableRequests().size();
+            } else if (response.getFailedRequests().size() > 0) {
+                int failedRequestsNumber = response.getFailedRequests().size();
                 log.error("Http Sink failed to write and will retry {} requests",
                     failedRequestsNumber);
                 numRecordsSendErrorsCounter.inc(failedRequestsNumber);

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

@@ -16,7 +16,6 @@
 import com.getindata.connectors.http.internal.HeaderPreprocessor;
 import com.getindata.connectors.http.internal.SinkHttpClient;
 import com.getindata.connectors.http.internal.SinkHttpClientResponse;
-import com.getindata.connectors.http.internal.SinkHttpClientResponse.ResponseItem;
 import com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants;
 import com.getindata.connectors.http.internal.sink.HttpSinkRequestEntry;
 import com.getindata.connectors.http.internal.status.ComposeHttpStatusCodeChecker;
@@ -98,7 +97,8 @@ private CompletableFuture<List<JavaNetHttpResponseWrapper>> submitRequests(
     private SinkHttpClientResponse prepareSinkHttpClientResponse(
         List<JavaNetHttpResponseWrapper> responses,
         String endpointUrl) {
-        var responseItems = new ArrayList<ResponseItem>();
+        var successfulResponses = new ArrayList<HttpRequest>();
+        var failedResponses = new ArrayList<HttpRequest>();
 
         for (var response : responses) {
             var sinkRequestEntry = response.getHttpRequest();
@@ -107,14 +107,19 @@ private SinkHttpClientResponse prepareSinkHttpClientResponse(
             httpPostRequestCallback.call(
                 optResponse.orElse(null), sinkRequestEntry, endpointUrl, headerMap);
 
-            // Empty indicates that HttpClient.sendAsync() failed with exception.
-            HttpResponseStatus status = optResponse.isEmpty()
-                ? HttpResponseStatus.FAILURE_RETRYABLE
-                : statusCodeChecker.checkStatus(optResponse.get().statusCode());
-            responseItems.add(new ResponseItem(sinkRequestEntry, status));
+            // TODO Add response processor here and orchestrate it with statusCodeChecker.
+            if (optResponse.isEmpty() ||
+                statusCodeChecker.checkStatus(optResponse.get().statusCode()).equals(
+                    HttpResponseStatus.FAILURE_RETRYABLE) ||
+                statusCodeChecker.checkStatus(optResponse.get().statusCode()).equals(
+                    HttpResponseStatus.FAILURE_NOT_RETRYABLE)) {
+                failedResponses.add(sinkRequestEntry);
+            } else {
+                successfulResponses.add(sinkRequestEntry);
+            }
         }
 
-        return new SinkHttpClientResponse(responseItems);
+        return new SinkHttpClientResponse(successfulResponses, failedResponses);
     }
 
     @VisibleForTesting

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

@@ -341,8 +341,7 @@ private void testSinkClientForConnection(
                 ).get();
 
             assertThat(response.getSuccessfulRequests()).isNotEmpty();
-            assertThat(response.getFailedNotRetryableRequests()).isEmpty();
-            assertThat(response.getFailedRetryableRequests()).isEmpty();
+            assertThat(response.getFailedRequests()).isEmpty();
         } catch (Exception e) {
             throw new RuntimeException(e);
         }