tamizhgeek commented on code in PR #255:
URL:
https://github.com/apache/flink-connector-aws/pull/255#discussion_r3815590733
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/reader/fanout/StreamConsumerRegistrar.java:
##########
@@ -143,6 +144,90 @@ public void deregisterStreamConsumer() {
}
}
+ /**
+ * Ensures an EFO consumer exists and is {@code ACTIVE}: waits out any
in-progress {@code
+ * DELETING}, then registers if missing. A concurrent {@link
ResourceInUseException} on register
+ * is treated as another caller winning the same race, not an error.
+ *
+ * @throws KinesisStreamsSourceException if it never reaches {@code
ACTIVE} within {@code
+ * maxAttempts}.
+ */
+ public static String ensureActiveConsumer(
+ StreamProxy kinesisStreamProxy,
+ String streamArn,
+ String consumerName,
+ Duration baseDelay,
+ Duration maxDelay,
+ int maxAttempts) {
+ for (int attempt = 1; ; attempt++) {
+ ConsumerStatus status;
+ String consumerArn;
+ try {
+ DescribeStreamConsumerResponse response =
+ kinesisStreamProxy.describeStreamConsumer(streamArn,
consumerName);
+ status = response.consumerDescription().consumerStatus();
+ consumerArn = response.consumerDescription().consumerARN();
+ } catch (ResourceNotFoundException e) {
+ status = null;
+ consumerArn = null;
+ }
+
+ if (status == ConsumerStatus.ACTIVE) {
+ return consumerArn;
+ }
+
+ if (status == null) {
+ try {
+ LOG.info("Registering stream consumer - {}::{}",
streamArn, consumerName);
+ kinesisStreamProxy.registerStreamConsumer(streamArn,
consumerName);
+ } catch (ResourceInUseException e) {
+ LOG.info(
+ "EFO consumer '{}' on stream {} appeared
concurrently while "
+ + "registering; waiting for it to become
ACTIVE.",
+ consumerName,
+ streamArn);
+ }
+ } else {
+ LOG.info(
+ "EFO consumer '{}' on stream {} is {}; waiting before
re-checking.",
+ consumerName,
+ streamArn,
+ status);
+ }
+
+ if (attempt >= maxAttempts) {
+ throw new KinesisStreamsSourceException(
+ "EFO consumer '"
+ + consumerName
+ + "' on stream "
+ + streamArn
+ + " did not become ACTIVE after "
+ + maxAttempts
+ + " attempts.",
+ null);
+ }
+
+ sleepUninterruptibly(computeBackoffMillis(baseDelay, maxDelay,
attempt));
+ }
+ }
+
+ /** Full-jitter exponential backoff: random duration between 0 and
min(base*2^(n-1), max). */
+ private static long computeBackoffMillis(Duration baseDelay, Duration
maxDelay, int attempt) {
+ long exponential = baseDelay.toMillis() * (1L << Math.min(attempt - 1,
20));
+ long capped = Math.min(exponential, maxDelay.toMillis());
+ return ThreadLocalRandom.current().nextLong(capped + 1);
+ }
+
+ private static void sleepUninterruptibly(long millis) {
Review Comment:
I think this should be called sleepInterruptibly?
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/reader/fanout/StreamConsumerRegistrar.java:
##########
@@ -143,6 +144,90 @@ public void deregisterStreamConsumer() {
}
}
+ /**
+ * Ensures an EFO consumer exists and is {@code ACTIVE}: waits out any
in-progress {@code
+ * DELETING}, then registers if missing. A concurrent {@link
ResourceInUseException} on register
+ * is treated as another caller winning the same race, not an error.
+ *
+ * @throws KinesisStreamsSourceException if it never reaches {@code
ACTIVE} within {@code
+ * maxAttempts}.
+ */
+ public static String ensureActiveConsumer(
+ StreamProxy kinesisStreamProxy,
+ String streamArn,
+ String consumerName,
+ Duration baseDelay,
+ Duration maxDelay,
+ int maxAttempts) {
+ for (int attempt = 1; ; attempt++) {
+ ConsumerStatus status;
+ String consumerArn;
+ try {
+ DescribeStreamConsumerResponse response =
+ kinesisStreamProxy.describeStreamConsumer(streamArn,
consumerName);
+ status = response.consumerDescription().consumerStatus();
+ consumerArn = response.consumerDescription().consumerARN();
+ } catch (ResourceNotFoundException e) {
+ status = null;
+ consumerArn = null;
+ }
+
+ if (status == ConsumerStatus.ACTIVE) {
+ return consumerArn;
+ }
+
+ if (status == null) {
+ try {
+ LOG.info("Registering stream consumer - {}::{}",
streamArn, consumerName);
+ kinesisStreamProxy.registerStreamConsumer(streamArn,
consumerName);
+ } catch (ResourceInUseException e) {
+ LOG.info(
+ "EFO consumer '{}' on stream {} appeared
concurrently while "
+ + "registering; waiting for it to become
ACTIVE.",
+ consumerName,
+ streamArn);
+ }
+ } else {
+ LOG.info(
+ "EFO consumer '{}' on stream {} is {}; waiting before
re-checking.",
+ consumerName,
+ streamArn,
+ status);
+ }
+
+ if (attempt >= maxAttempts) {
+ throw new KinesisStreamsSourceException(
+ "EFO consumer '"
+ + consumerName
+ + "' on stream "
+ + streamArn
+ + " did not become ACTIVE after "
+ + maxAttempts
+ + " attempts.",
+ null);
+ }
+
+ sleepUninterruptibly(computeBackoffMillis(baseDelay, maxDelay,
attempt));
+ }
+ }
+
+ /** Full-jitter exponential backoff: random duration between 0 and
min(base*2^(n-1), max). */
Review Comment:
I would use
software.amazon.awssdk.core.retry.backoff.FullJitterBackoffStrategy instead of
building our own backoff strategy
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/reader/fanout/StreamConsumerRegistrar.java:
##########
@@ -143,6 +144,90 @@ public void deregisterStreamConsumer() {
}
}
+ /**
+ * Ensures an EFO consumer exists and is {@code ACTIVE}: waits out any
in-progress {@code
+ * DELETING}, then registers if missing. A concurrent {@link
ResourceInUseException} on register
+ * is treated as another caller winning the same race, not an error.
+ *
+ * @throws KinesisStreamsSourceException if it never reaches {@code
ACTIVE} within {@code
+ * maxAttempts}.
+ */
+ public static String ensureActiveConsumer(
+ StreamProxy kinesisStreamProxy,
+ String streamArn,
+ String consumerName,
+ Duration baseDelay,
+ Duration maxDelay,
+ int maxAttempts) {
+ for (int attempt = 1; ; attempt++) {
+ ConsumerStatus status;
+ String consumerArn;
+ try {
+ DescribeStreamConsumerResponse response =
+ kinesisStreamProxy.describeStreamConsumer(streamArn,
consumerName);
+ status = response.consumerDescription().consumerStatus();
+ consumerArn = response.consumerDescription().consumerARN();
+ } catch (ResourceNotFoundException e) {
+ status = null;
+ consumerArn = null;
+ }
+
+ if (status == ConsumerStatus.ACTIVE) {
+ return consumerArn;
+ }
+
+ if (status == null) {
+ try {
+ LOG.info("Registering stream consumer - {}::{}",
streamArn, consumerName);
+ kinesisStreamProxy.registerStreamConsumer(streamArn,
consumerName);
Review Comment:
Its a bit weird that we try to register the consumer here, but then fail it
at line 198 without even checking if this attempt has succeeded. I would either
move the check for attempts >= maxAttempts to the top of the method, or once
registered here, check if its successful and return the consumer ARN.
--
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]
