bharathgunapati commented on code in PR #54:
URL:
https://github.com/apache/flink-connector-http/pull/54#discussion_r4199703936
##########
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:
Makes sense. Added prepareHeaderMap(...) (and getRequestMode()) on
HttpSinkConfig, so callers no longer dig headers out of the raw properties map.
We still keep properties on the sink config because open-ended keys beyond
headers still live there (TLS, legacy error codes, batch size, logger, etc.).
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
