Re: [PR] [FLINK-40277][connector-http] Add HTTP sink status-code handling and retries [flink-connector-http]

["davidradl (via GitHub)" <gi...@apache.org>]

davidradl commented on code in PR #54:
URL: https://github.com/apache/flink-connector-http/pull/54#discussion_r4197893195


##########
flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClient.java:
##########
@@ -52,51 +53,72 @@ public class JavaNetSinkHttpClient implements SinkHttpClient {
 
     private final Map<String, String> headerMap;
 
-    private final HttpStatusCodeChecker statusCodeChecker;
+    private final HttpSinkResponseClassifier responseClassifier;
 
     private final HttpPostRequestCallback<HttpRequest> httpPostRequestCallback;
 
     private final RequestSubmitter requestSubmitter;
 
     private final HttpLogger httpLogger;
 
-    public JavaNetSinkHttpClient(
-            Properties properties,
-            HttpPostRequestCallback<HttpRequest> httpPostRequestCallback,
+    private final HttpSinkClientWithRetry httpClientWithRetry;
+
+    public static SinkHttpClientBuilder builder() {
+        return new SinkHttpClientBuilder() {
+            @Override
+            public SinkHttpClient build(SinkHttpClientContext context) {
+                return new JavaNetSinkHttpClient(
+                        context.getSinkConfig(),
+                        context.getHeaderPreprocessor(),
+                        createRequestSubmitterFactory(context));
+            }
+        };
+    }
+
+    JavaNetSinkHttpClient(
+            HttpSinkConfig sinkConfig,
             HeaderPreprocessor headerPreprocessor,
             RequestSubmitterFactory requestSubmitterFactory) {
 
-        this.httpPostRequestCallback = httpPostRequestCallback;
+        var properties = sinkConfig.getProperties();
+        this.httpPostRequestCallback = sinkConfig.getHttpPostRequestCallback();
         this.headerMap =
                 HttpHeaderUtils.prepareHeaderMap(
                         HttpConnectorConfigConstants.SINK_HEADER_PREFIX,
                         properties,
                         headerPreprocessor);
 
-        // TODO Inject this via constructor when implementing a response processor.
-        //  Processor will be injected and it will wrap statusChecker implementation.
-        ComposeHttpStatusCodeCheckerConfig checkerConfig =
-                ComposeHttpStatusCodeCheckerConfig.builder()
-                        .properties(properties)
-                        .includeListPrefix(
-                                HttpConnectorConfigConstants.HTTP_ERROR_SINK_CODE_INCLUDE_LIST)
-                        .errorCodePrefix(HttpConnectorConfigConstants.HTTP_ERROR_SINK_CODES_LIST)
-                        .build();
-
-        this.statusCodeChecker = new ComposeHttpStatusCodeChecker(checkerConfig);
+        this.responseClassifier = new HttpSinkResponseClassifier(sinkConfig);
 
         this.headersAndValues = HttpHeaderUtils.toHeaderAndValueArray(this.headerMap);
         this.requestSubmitter =
-                requestSubmitterFactory.createSubmitter(properties, headersAndValues);

Review Comment:
   It looks odd that we were passing the properties and now we pass the sinkconfig . But the properties are obtained from the sinkconfig. Also the header values are derived from the properties.  
   
   Do we need to pass the properties into sinkconfig? It seems the only reason it has it is to be able to get the header information out. Can we have a getter for the header information? Instead of having to store the whole properties?  



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org