[jira] [Commented] (FLINK-38814) Kinesis Source in EFO - idle consumer on subscription failure

["Juliusz Nadberezny (Jira)" <ji...@apache.org>]

    [ 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)