dannycranmer commented on code in PR #39:
URL:
https://github.com/apache/flink-connector-aws/pull/39#discussion_r1054235701
##########
flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/publisher/fanout/StreamConsumerRegistrar.java:
##########
@@ -77,8 +79,8 @@ public StreamConsumerRegistrar(
* @throws InterruptedException
*/
public String registerStreamConsumer(final String stream, final String
streamConsumerName)
- throws ExecutionException, InterruptedException {
- LOG.debug("Registering stream consumer - {}::{}", stream,
streamConsumerName);
+ throws Exception {
Review Comment:
Why did you make this change? This will possible change the error handling
behaviour
##########
flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/publisher/fanout/StreamConsumerRegistrar.java:
##########
@@ -90,7 +92,7 @@ public String registerStreamConsumer(final String stream,
final String streamCon
kinesisProxyV2Interface.describeStreamSummary(stream);
String streamArn =
describeStreamSummaryResponse.streamDescriptionSummary().streamARN();
- LOG.debug("Found stream ARN - {}", streamArn);
+ LOG.warn("Found stream ARN - {}", streamArn);
Review Comment:
Why is this a warning?
##########
flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/publisher/fanout/StreamConsumerRegistrar.java:
##########
@@ -77,8 +79,8 @@ public StreamConsumerRegistrar(
* @throws InterruptedException
*/
public String registerStreamConsumer(final String stream, final String
streamConsumerName)
- throws ExecutionException, InterruptedException {
- LOG.debug("Registering stream consumer - {}::{}", stream,
streamConsumerName);
+ throws Exception {
+ LOG.warn("Registering stream consumer - {}::{}", stream,
streamConsumerName);
Review Comment:
Why is this a warning?
##########
flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/publisher/fanout/StreamConsumerRegistrar.java:
##########
@@ -243,28 +244,28 @@ private void waitForConsumerToDeregister(
}
private Optional<DescribeStreamConsumerResponse> describeStreamConsumer(
- final String streamArn, final String streamConsumerName)
- throws InterruptedException, ExecutionException {
+ final String streamArn, final String streamConsumerName) throws
Exception {
return describeStreamConsumer(
() ->
kinesisProxyV2Interface.describeStreamConsumer(
streamArn, streamConsumerName));
}
private Optional<DescribeStreamConsumerResponse> describeStreamConsumer(
- final String streamConsumerArn) throws InterruptedException,
ExecutionException {
+ final String streamConsumerArn) throws Exception {
return describeStreamConsumer(
() ->
kinesisProxyV2Interface.describeStreamConsumer(streamConsumerArn));
}
private Optional<DescribeStreamConsumerResponse> describeStreamConsumer(
- final ResponseSupplier<DescribeStreamConsumerResponse>
responseSupplier)
- throws InterruptedException, ExecutionException {
+ final Callable<DescribeStreamConsumerResponse> responseSupplier)
throws Exception {
DescribeStreamConsumerResponse response;
try {
- response = responseSupplier.get();
- } catch (ExecutionException ex) {
+ response = responseSupplier.call();
+ } catch (Exception ex) {
+ LOG.warn("describeStreamConsumer caught ExecutionException: {}",
ex);
Review Comment:
`caught ExecutionException` not necessarily now you are catching
`Exception`. This change is slightly worrying, what happens now if an
`InterruptedException` is thrown?
##########
flink-connector-aws-base/src/main/java/org/apache/flink/connector/aws/util/AWSAsyncSinkUtil.java:
##########
@@ -161,4 +163,47 @@ S createAwsAsyncClient(
.region(getRegion(configProps))
.build();
}
+
+ /**
+ * @param configProps configuration properties
Review Comment:
This javadoc is missing a comment.
##########
flink-connector-kinesis/pom.xml:
##########
@@ -256,6 +256,11 @@ under the License.
<artifactId>netty-nio-client</artifactId>
</dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>apache-client</artifactId>
+ </dependency>
+
Review Comment:
Why is this needed explicitly in here and `aws-base`?
##########
flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxySyncV2.java:
##########
@@ -114,6 +104,7 @@ public DescribeStreamConsumerResponse
describeStreamConsumer(
.streamARN(streamArn)
.consumerName(consumerName)
.build();
+ LOG.warn("describeStreamConsumer with arn: {}, consumerName: {}",
streamArn, consumerName);
Review Comment:
Why is this a warning? (did you leave dev debug logs in?)
##########
flink-connector-aws-base/src/main/java/org/apache/flink/connector/aws/util/AWSAsyncSinkUtil.java:
##########
@@ -161,4 +163,47 @@ S createAwsAsyncClient(
.region(getRegion(configProps))
.build();
}
+
+ /**
+ * @param configProps configuration properties
+ * @param httpClient the underlying HTTP client used to talk to AWS
+ * @return a new AWS Sync Client
+ */
+ public static <
+ S extends SdkClient,
+ T extends
+ AwsSyncClientBuilder<? extends T, S> &
AwsClientBuilder<? extends T, S>>
+ S createAwsSyncClient(
+ final Properties configProps,
+ final SdkHttpClient httpClient,
+ final T clientBuilder,
+ final String awsUserAgentPrefixFormat,
+ final String awsClientUserAgentPrefix) {
+ SdkClientConfiguration clientConfiguration =
SdkClientConfiguration.builder().build();
+
+ String flinkUserAgentPrefix =
+
Optional.ofNullable(configProps.getProperty(awsClientUserAgentPrefix))
+ .orElse(
+ formatFlinkUserAgentPrefix(
+ awsUserAgentPrefixFormat +
V2_USER_AGENT_SUFFIX));
+
+ final ClientOverrideConfiguration overrideConfiguration =
+ createClientOverrideConfiguration(
+ clientConfiguration,
+ ClientOverrideConfiguration.builder(),
+ flinkUserAgentPrefix);
+
+ if (configProps.containsKey(AWSConfigConstants.AWS_ENDPOINT)) {
+ final URI endpointOverride =
+
URI.create(configProps.getProperty(AWSConfigConstants.AWS_ENDPOINT));
+ clientBuilder.endpointOverride(endpointOverride);
+ }
+
Review Comment:
This code is duplicated in this file, can you extract to a method please
##########
flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyAsyncV2Test.java:
##########
@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.kinesis.proxy;
+
+import
org.apache.flink.streaming.connectors.kinesis.internals.publisher.fanout.FanOutRecordPublisherConfiguration;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import software.amazon.awssdk.http.SdkHttpClient;
+import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
+import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.model.SubscribeToShardRequest;
+import
software.amazon.awssdk.services.kinesis.model.SubscribeToShardResponseHandler;
+
+import java.util.Properties;
+
+import static java.util.Collections.emptyList;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.DEREGISTER_STREAM_BACKOFF_BASE;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.DEREGISTER_STREAM_BACKOFF_EXPONENTIAL_CONSTANT;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.DEREGISTER_STREAM_BACKOFF_MAX;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.DESCRIBE_STREAM_CONSUMER_BACKOFF_BASE;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.DESCRIBE_STREAM_CONSUMER_BACKOFF_EXPONENTIAL_CONSTANT;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.DESCRIBE_STREAM_CONSUMER_BACKOFF_MAX;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.EFO_CONSUMER_NAME;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.RECORD_PUBLISHER_TYPE;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.REGISTER_STREAM_BACKOFF_BASE;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.REGISTER_STREAM_BACKOFF_EXPONENTIAL_CONSTANT;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.REGISTER_STREAM_BACKOFF_MAX;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.RecordPublisherType.EFO;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_BASE;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_EXPONENTIAL_CONSTANT;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.STREAM_DESCRIBE_BACKOFF_MAX;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.STREAM_DESCRIBE_RETRIES;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.SUBSCRIBE_TO_SHARD_BACKOFF_BASE;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.SUBSCRIBE_TO_SHARD_BACKOFF_EXPONENTIAL_CONSTANT;
+import static
org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.SUBSCRIBE_TO_SHARD_BACKOFF_MAX;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
Review Comment:
Mockito is [banned from
use](https://flink.apache.org/contributing/code-style-and-quality-common.html#avoid-mockito---use-reusable-test-implementations)
in Flink. I know we do already use it in places but we should not use for new
tests. Can we refactor to reusable test implementations?
##########
flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxySyncV2.java:
##########
@@ -148,11 +139,10 @@ public RegisterStreamConsumerResponse
registerStreamConsumer(
.consumerName(consumerName)
.build();
+ LOG.warn("registerStreamConsumer with arn: {}, consumerName: {}",
streamArn, consumerName);
Review Comment:
Why is this a warning?
--
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]
