nuno-c-afonso commented on a change in pull request #17345:
URL: https://github.com/apache/flink/pull/17345#discussion_r756798954
##########
File path:
flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AwsV2Util.java
##########
@@ -66,254 +36,23 @@
/** Utility methods specific to Amazon Web Service SDK v2.x. */
@Internal
public class AwsV2Util {
-
- private static final int INITIAL_WINDOW_SIZE_BYTES = 512 * 1024; // 512 KB
- private static final Duration HEALTH_CHECK_PING_PERIOD =
Duration.ofSeconds(60);
- private static final Duration CONNECTION_ACQUISITION_TIMEOUT =
Duration.ofSeconds(60);
-
- /**
- * Creates an Amazon Kinesis Async Client from the provided properties.
Configuration is copied
- * from AWS SDK v1 configuration class as per: -
- *
https://github.com/aws/aws-sdk-java-v2/blob/2.13.52/docs/LaunchChangelog.md#134-client-override-retry-configuration
- *
- * @param configProps configuration properties
- * @param clientConfiguration the AWS SDK v1.X config ported to V2 to
instantiate the client
- * @param httpClient the underlying HTTP client used to talk to Kinesis
- * @return a new Amazon Kinesis Client
- */
- public static KinesisAsyncClient createKinesisAsyncClient(
- final Properties configProps,
- final ClientConfiguration clientConfiguration,
- final SdkAsyncHttpClient httpClient) {
- final ClientOverrideConfiguration overrideConfiguration =
- createClientOverrideConfiguration(
- clientConfiguration,
ClientOverrideConfiguration.builder());
- final KinesisAsyncClientBuilder clientBuilder =
KinesisAsyncClient.builder();
-
- return createKinesisAsyncClient(
- configProps, clientBuilder, httpClient, overrideConfiguration);
- }
-
- public static SdkAsyncHttpClient createHttpClient(
- final ClientConfiguration config,
- final NettyNioAsyncHttpClient.Builder httpClientBuilder,
- final Properties consumerConfig) {
-
- int maxConcurrency =
-
Optional.ofNullable(consumerConfig.getProperty(EFO_HTTP_CLIENT_MAX_CONCURRENCY))
- .map(Integer::parseInt)
- .orElse(DEFAULT_EFO_HTTP_CLIENT_MAX_CONCURRENCY);
-
- Duration readTimeout =
-
Optional.ofNullable(consumerConfig.getProperty(EFO_HTTP_CLIENT_READ_TIMEOUT_MILLIS))
- .map(Integer::parseInt)
- .map(Duration::ofMillis)
- .orElse(DEFAULT_EFO_HTTP_CLIENT_READ_TIMEOUT);
-
- httpClientBuilder
- .maxConcurrency(maxConcurrency)
-
.connectionTimeout(Duration.ofMillis(config.getConnectionTimeout()))
- .readTimeout(readTimeout)
- .tcpKeepAlive(config.useTcpKeepAlive())
- .writeTimeout(Duration.ofMillis(config.getSocketTimeout()))
-
.connectionMaxIdleTime(Duration.ofMillis(config.getConnectionMaxIdleMillis()))
- .useIdleConnectionReaper(config.useReaper())
- .protocol(Protocol.HTTP2)
- .connectionAcquisitionTimeout(CONNECTION_ACQUISITION_TIMEOUT)
- .http2Configuration(
- Http2Configuration.builder()
-
.healthCheckPingPeriod(HEALTH_CHECK_PING_PERIOD)
- .initialWindowSize(INITIAL_WINDOW_SIZE_BYTES)
- .build());
Review comment:
In `AWSKinesisDataStreamsUtil.java`, the missing assignments are done in
`NettyNioAsyncHttpClient.Builder::buildWithDefaults` when extracting the
properties from the provided `AttributeMap`.
--
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]
