diff --git a/amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShardConsumer.java b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShardConsumer.java index 96261131..16dac30b 100644 --- a/amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShardConsumer.java +++ b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShardConsumer.java @@ -179,8 +179,18 @@ public class ShardConsumer { // Task rejection during the subscribe() call will not be propagated back as it not executed // in the context of the Scheduler thread. Hence we should not assume the subscription will // always be successful. + // But if subscription was not successful, then it will recover + // during healthCheck which will restart subscription. + // From Shardconsumer point of view, initialization after the below subscribe call + // is complete subscribe(); needsInitialization = false; + // Initialization is complete, return now, because we dont need to do + // initializeComplete anymore. ShardConsumer is in ProcessingState now and any further activity + // will be driven by publisher pushing data to subscriber which invokes handleInput + // and that triggers ProcessTask. Scheduler is only meant to do health-checks + // to ensure the consumer is not stuck for any reason and to do shutdown handling. + return; } } stateChangeFuture = initializeComplete();