[
https://issues.apache.org/jira/browse/FLINK-38814?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18046144#comment-18046144
]
Juliusz Nadberezny commented on FLINK-38814:
--------------------------------------------
[~Hong Teoh] Could you please take a look at this PR when you have a moment? I
think you have the best knowledge in this area :)
> Kinesis Source in EFO - idle consumer on subscription failure
> -------------------------------------------------------------
>
> Key: FLINK-38814
> URL: https://issues.apache.org/jira/browse/FLINK-38814
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kinesis
> Affects Versions: aws-connector-5.0.0
> Reporter: Juliusz Nadberezny
> Priority: Major
>
> The {{FanOutKinesisShardSubscription}} can enter a zombie state where the
> connector appears active but consumes no data. This occurs when a
> subscription attempt fails (due to {{{}ResourceInUseException{}}}),
> triggering a silent {{NPE}} in the background activation thread.
> The issue lies in the {{.exceptionally}} block inside
> {{{}activateSubscription(){}}}.
> When Kinesis throws a {{{}ResourceInUseException{}}}, the code explicitly
> tells the {{waitForSubscriptionLatch}} to release. The waiting thread wakes
> up, {*}assuming the subscription was successful{*}, and attempts to use it.
> *PR with the fix:*
> https://github.com/apache/flink-connector-aws/pull/224
> *The code path to failure:*
> 1. Trigger
> {{kinesis.subscribeToShard}} fails with {{ResourceInUseException}}
> 2. The Trap
> The error handler catches it
> {code:java}
> if (ExceptionUtils.findThrowable(throwable,
> ResourceInUseException.class).isPresent()) {
> waitForSubscriptionLatch.countDown(); // <--- TRAP: Signals "Success" to
> the waiter
> return null;
> } {code}
> 3. The Crash
> The background thread (inside {{{}CompletableFuture.runAsync{}}}) wakes up
> because the latch opened. It proceeds to the "success" block:
> {code:java}
> if (waitForSubscriptionLatch.await(...)) {
> subscriptionActive.set(true); // 1. Marks connector as ACTIVE
> shardSubscriber.requestRecords(); // 2. Calls method on subscriber
> } {code}
> 4. The NPE
> Inside {{{}shardSubscriber.requestRecords(){}}}:
> {code:java}
> public void requestRecords() {
> subscription.request(1); // <--- CRASH: 'subscription' is NULL
> } {code}
> *How to observe it in logs:*
> {code:java}
> if (ExceptionUtils.findThrowable(
> throwable, ResourceInUseException.class)
> .isPresent()) {
> LOG.warn("DEBUG: ResourceInUseException caught. Counting down latch
> (Original Logic triggers here).");
> waitForSubscriptionLatch.countDown();
> return null;
> } {code}
> {code:java}
> try { shardSubscriber.requestRecords(); LOG.info("DEBUG: [AsyncThread]
> requestRecords() success.");} catch (NullPointerException e) {
> LOG.error("DEBUG: [AsyncThread] CRITICAL: Caught NPE in requestRecords!
> Subscription object is likely missing.", e); terminateSubscription(e);}
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
