Refactoring

Author: Grzegorz Kołakowski

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

@@ -46,18 +46,18 @@ public class HttpSinkWriter<InputT> extends AsyncSinkWriter<InputT, HttpSinkRequ
     private final Counter numRecordsSendErrorsCounter;
 
     public HttpSinkWriter(
-        ElementConverter<InputT, HttpSinkRequestEntry> elementConverter,
-        Sink.InitContext context,
-        int maxBatchSize,
-        int maxInFlightRequests,
-        int maxBufferedRequests,
-        long maxBatchSizeInBytes,
-        long maxTimeInBufferMS,
-        long maxRecordSizeInBytes,
-        String endpointUrl,
-        SinkHttpClient sinkHttpClient,
-        Collection<BufferedRequestState<HttpSinkRequestEntry>> bufferedRequestStates,
-        Properties properties) {
+            ElementConverter<InputT, HttpSinkRequestEntry> elementConverter,
+            Sink.InitContext context,
+            int maxBatchSize,
+            int maxInFlightRequests,
+            int maxBufferedRequests,
+            long maxBatchSizeInBytes,
+            long maxTimeInBufferMS,
+            long maxRecordSizeInBytes,
+            String endpointUrl,
+            SinkHttpClient sinkHttpClient,
+            Collection<BufferedRequestState<HttpSinkRequestEntry>> bufferedRequestStates,
+            Properties properties) {
 
         super(elementConverter, context, maxBatchSize, maxInFlightRequests, maxBufferedRequests,
             maxBatchSizeInBytes, maxTimeInBufferMS, maxRecordSizeInBytes, bufferedRequestStates);
@@ -82,8 +82,8 @@ public HttpSinkWriter(
     // TODO: Reintroduce retries by adding backoff policy
     @Override
     protected void submitRequestEntries(
-        List<HttpSinkRequestEntry> requestEntries,
-        Consumer<List<HttpSinkRequestEntry>> requestResult) {
+            List<HttpSinkRequestEntry> requestEntries,
+            Consumer<List<HttpSinkRequestEntry>> requestResult) {
         var future = sinkHttpClient.putRequests(requestEntries, endpointUrl);
         future.whenCompleteAsync((response, err) -> {
             if (err != null) {

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

@@ -110,14 +110,12 @@ private SinkHttpClientResponse prepareSinkHttpClientResponse(
                 optResponse.orElse(null), sinkRequestEntry, endpointUrl, headerMap);
 
             // 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 {
+            if (optResponse.isPresent() &&
+                statusCodeChecker.checkStatus(optResponse.get().statusCode())
+                    .equals(HttpResponseStatus.SUCCESS)) {
                 successfulResponses.add(sinkRequestEntry);
+            } else {
+                failedResponses.add(sinkRequestEntry);
             }
         }
 

src/main/java/com/getindata/connectors/http/internal/status/ComposeHttpStatusCodeChecker.java

@@ -88,25 +88,25 @@ private Optional<Predicate<Integer>> prepareErrorCodes(String statusCodesStr) {
         return Arrays.stream(statusCodesStr.split(HttpConnectorConfigConstants.PROP_DELIM))
             .filter(code -> !isNullOrWhitespaceOnly(code))
             .map(code -> code.toUpperCase().trim())
-            .map(codeStr -> {
-                Preconditions.checkArgument(
-                    codeStr.length() == 3,
-                    "Status code should contain three characters. Provided [%s]",
-                    codeStr);
-
-                // at this point we have trim, upper case 3 character status code.
-                if (isTypeCode(codeStr)) {
-                    int code = Integer.parseInt(codeStr.replace("X", ""));
-                    return new TypeStatusCodeCheckerPredicate(
-                        HttpResponseCodeType.getByCode(code));
-                } else {
-                    return new SingleValueHttpStatusCodeCheckerPredicate(
-                        Integer.parseInt(codeStr));
-                }
-            })
+            .map(this::prepareErrorCode)
             .reduce(Predicate::or);
     }
 
+    private Predicate<Integer> prepareErrorCode(String codeString) {
+        Preconditions.checkArgument(
+            codeString.length() == 3,
+            "Status code should contain three characters. Provided [%s]",
+            codeString);
+
+        // at this point we have trim, upper case 3 character status code.
+        if (isTypeCode(codeString)) {
+            int code = Integer.parseInt(codeString.replace("X", ""));
+            return new TypeStatusCodeCheckerPredicate(HttpResponseCodeType.getByCode(code));
+        } else {
+            return new SingleValueHttpStatusCodeCheckerPredicate(Integer.parseInt(codeString));
+        }
+    }
+
     /**
      * This method checks if "code" param matches "digit + XX" mask. This method expects that
      * provided string will be 3 elements long, trim and upper case.