jlalwani-amazon commented on code in PR #243: URL: https://github.com/apache/flink-connector-aws/pull/243#discussion_r3151464974
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/reader/fanout/FanOutKinesisShardSubscription.java:
##########
@@ -237,12 +236,26 @@ public SubscribeToShardEvent nextEvent() {
"Subscription encountered unrecoverable exception.",
throwable);
}
- if (!subscriptionActive.get()) {
+ if (shardSubscriber == null || shardSubscriber.getSubscriptionState()
== SubscriptionState.NOT_STARTED) {
Review Comment:
This code block will be cleaner if you use `switch..case` and `Optional`
instead of `if/else` Also, are you sure that you always want `eventQueue.poll`
to be the fallback? What if a new enum is added to `SubscriptionState`? Should
this code fail if a new enum is added?
I would make this
```
SubscriptionState state = Optional.ofNullable(shardSubscriber)
.map(FanOutShardSubscriber::getSubscriptionState)
.orElse(SubscriptionState.NOT_STARTED);
switch (state) {
case NOT_STARTED:
LOG.debug("Subscription to shard {} for consumer {} is not yet
active. Skipping.",
shardId, consumerArn);
return null;
case COMPLETED:
if (shardSubscriber.reachedShardEnd()) {
LOG.info("Subscription reached SHARD_END for shard {} for
consumer {}.",
shardId, consumerArn);
return null;
}
LOG.info("Subscription expired to shard {} for consumer {}.
Restarting.",
shardId, consumerArn);
activateSubscription();
return null;
case SUBSCRIBED:
return eventQueue.poll();
default:
throw new IllegalStateException("Unknown subscription state: "
+ state);
}
```
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/reader/fanout/FanOutKinesisShardSubscription.java:
##########
@@ -254,26 +267,34 @@ public SubscribeToShardEvent nextEvent() {
*/
private class FanOutShardSubscriber implements
Subscriber<SubscribeToShardEventStream> {
private final CountDownLatch subscriptionLatch;
-
private Subscription subscription;
+ private final AtomicReference<SubscriptionState> subscriptionState =
+ new AtomicReference<>(SubscriptionState.NOT_STARTED);
+ private final AtomicBoolean reachedShardEnd = new AtomicBoolean(false);
Review Comment:
To be consistent, we should have a `isReachedShardEnd` just like we have a
`getSubscriptionState` function
--
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]
