(nifi) branch main updated: NIFI-15753 Improved ConsumeKinesis Polling checks for pending fetches (#11050)

[exceptionfactory]

This is an automated email from the ASF dual-hosted git repository.

exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git


The following commit(s) were added to refs/heads/main by this push:
     new e83b1c2de68 NIFI-15753 Improved ConsumeKinesis Polling checks for 
pending fetches (#11050)
e83b1c2de68 is described below

commit e83b1c2de68fbe3513bf9db5db3806b02f1917a5
Author: Pierre Villard <pierre.villard...@gmail.com>
AuthorDate: Wed Apr 1 22:41:12 2026 +0200

    NIFI-15753 Improved ConsumeKinesis Polling checks for pending fetches 
(#11050)
    
    - Added timeouts to PollingKinesisClientTest methods
    
    Signed-off-by: David Handermann <exceptionfact...@apache.org>
---
 .../apache/nifi/processors/aws/kinesis/PollingKinesisClient.java | 5 ++++-
 .../nifi/processors/aws/kinesis/PollingKinesisClientTest.java    | 9 +++------
 2 files changed, 7 insertions(+), 7 deletions(-)

diff --git 
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/PollingKinesisClient.java
 
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/PollingKinesisClient.java
index 5e9cdfc0bea..8cb21e46b95 100644
--- 
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/PollingKinesisClient.java
+++ 
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/PollingKinesisClient.java
@@ -102,6 +102,9 @@ final class PollingKinesisClient extends 
KinesisConsumerClient {
 
     @Override
     boolean hasPendingFetches() {
+        if (fetchExecutor.isShutdown()) {
+            return false;
+        }
         if (hasQueuedResults()) {
             return true;
         }
@@ -328,7 +331,7 @@ final class PollingKinesisClient extends 
KinesisConsumerClient {
 
     private boolean enqueueIfActive(final String shardId, final 
PollingShardState state, final ShardFetchResult result) {
         synchronized (getShardLock(shardId)) {
-            if (state.isResetRequested()) {
+            if (state.isResetRequested() || state.isStopped()) {
                 return false;
             }
             enqueueResult(result);
diff --git 
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/test/java/org/apache/nifi/processors/aws/kinesis/PollingKinesisClientTest.java
 
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/test/java/org/apache/nifi/processors/aws/kinesis/PollingKinesisClientTest.java
index c663c6f4f1b..d89ad1bb0b2 100644
--- 
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/test/java/org/apache/nifi/processors/aws/kinesis/PollingKinesisClientTest.java
+++ 
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/test/java/org/apache/nifi/processors/aws/kinesis/PollingKinesisClientTest.java
@@ -246,9 +246,8 @@ class PollingKinesisClientTest {
         while (System.nanoTime() < replayQueuedDeadline && 
(getShardIteratorCallCount.get() < 2 || getRecordsCallCount.get() < 4)) {
             Thread.sleep(20);
         }
-        Thread.sleep(100);
 
-        final ShardFetchResult firstAfterRecovery = 
consumer.pollShardResult("shard-1");
+        final ShardFetchResult firstAfterRecovery = consumer.pollAnyResult(5, 
TimeUnit.SECONDS);
         assertNotNull(firstAfterRecovery, "Queue must contain results after 
iterator recovery");
         assertEquals(new BigInteger("200"), 
firstAfterRecovery.firstSequenceNumber(),
                 "Replayed checkpoint data must be delivered before any stale 
newer batch from the same shard");
@@ -408,9 +407,8 @@ class PollingKinesisClientTest {
         while (getShardIteratorCallCount.get() < 2 && System.nanoTime() < 
resetDeadline) {
             Thread.sleep(20);
         }
-        Thread.sleep(100);
 
-        final ShardFetchResult firstAfterRollback = 
consumer.pollShardResult("shard-1");
+        final ShardFetchResult firstAfterRollback = consumer.pollAnyResult(5, 
TimeUnit.SECONDS);
         assertNotNull(firstAfterRollback, "Queue must contain results after 
rollback and re-fetch");
         assertEquals(new BigInteger("500"), 
firstAfterRollback.firstSequenceNumber(),
                 "First result after rollback must be re-fetched data, not a 
stale pre-rollback result");
@@ -552,9 +550,8 @@ class PollingKinesisClientTest {
         while (getShardIteratorCallCount.get() < 2 && System.nanoTime() < 
resetDeadline) {
             Thread.sleep(20);
         }
-        Thread.sleep(100);
 
-        final ShardFetchResult firstAfterRollback = 
consumer.pollShardResult("shard-1");
+        final ShardFetchResult firstAfterRollback = consumer.pollAnyResult(5, 
TimeUnit.SECONDS);
         assertNotNull(firstAfterRollback, "Queue must contain results after 
rollback and re-fetch");
         assertEquals(new BigInteger("800"), 
firstAfterRollback.firstSequenceNumber(),
                 "First result after rollback must be re-fetched data (800), 
not the stale data (500) that was returned from GetRecords during the 
rollback");