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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
