You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ex...@apache.org on 2026/04/03 20:21:29 UTC
(nifi) branch main updated: NIFI-15792 Added background message consumption in ConsumeKinesis (#11102)
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 05ea8b959f8 NIFI-15792 Added background message consumption in ConsumeKinesis (#11102)
05ea8b959f8 is described below
commit 05ea8b959f8e2cf4d78f92942b8efb7fc163cddf
Author: Mark Payne <ma...@hotmail.com>
AuthorDate: Fri Apr 3 16:21:23 2026 -0400
NIFI-15792 Added background message consumption in ConsumeKinesis (#11102)
Trigger background consumption of messages in ConsumeKinesis in such a way that data is pulled immediately instead of waiting on session commit to complete; when processor is stopped we also quickly bail out of our data gathering loop.
Also found in Stateless that ConsumeKinesis errors when needing to create the DynamoDB table because stateless defaults to a 10 second timeout for @OnScheduled while standard defaults to a 1 minute timeout so updated to 1 min for stateless in order to make them consistent
Signed-off-by: David Handermann <ex...@apache.org>
---
.../processors/aws/kinesis/ConsumeKinesis.java | 2 +-
.../aws/kinesis/EnhancedFanOutClient.java | 6 +++++
.../aws/kinesis/KinesisConsumerClientTest.java | 29 ++++++++++++++++++++++
.../PropertiesFileEngineConfigurationParser.java | 2 +-
.../engine/StatelessEngineConfiguration.java | 2 +-
5 files changed, 38 insertions(+), 3 deletions(-)
diff --git a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/ConsumeKinesis.java b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/ConsumeKinesis.java
index aab3b702453..bf3f6afc9d3 100644
--- a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/ConsumeKinesis.java
+++ b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/ConsumeKinesis.java
@@ -708,7 +708,7 @@ public class ConsumeKinesis extends AbstractProcessor {
final long startNanos = System.nanoTime();
long estimatedBytes = 0;
- while (System.nanoTime() < startNanos + maxBatchNanos && estimatedBytes < maxBatchBytes) {
+ while (isScheduled() && System.nanoTime() < startNanos + maxBatchNanos && estimatedBytes < maxBatchBytes) {
final List<String> readyShards = consumerClient.getShardIdsWithResults();
if (readyShards.isEmpty()) {
if (!consumerClient.hasPendingFetches()) {
diff --git a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/EnhancedFanOutClient.java b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/EnhancedFanOutClient.java
index bf0373561eb..4db3bc9021c 100644
--- a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/EnhancedFanOutClient.java
+++ b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/EnhancedFanOutClient.java
@@ -146,6 +146,11 @@ final class EnhancedFanOutClient extends KinesisConsumerClient {
return total;
}
+ @Override
+ protected void onResultPolled() {
+ resumePausedConsumers();
+ }
+
@Override
void acknowledgeResults(final List<ShardFetchResult> results) {
resumePausedConsumers();
@@ -355,6 +360,7 @@ final class EnhancedFanOutClient extends KinesisConsumerClient {
private volatile long lastSubscribeAttemptNanos;
private volatile BigInteger lastQueuedSequenceNumber;
private volatile boolean shardNotFound;
+
ShardConsumer(final String shardId, final Consumer<ShardFetchResult> resultSink,
final Queue<ShardConsumer> pausedConsumers, final ComponentLog consumerLogger) {
this.shardId = shardId;
diff --git a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/test/java/org/apache/nifi/processors/aws/kinesis/KinesisConsumerClientTest.java b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/test/java/org/apache/nifi/processors/aws/kinesis/KinesisConsumerClientTest.java
index 6f6f124a5f6..66fbee0334f 100644
--- a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/test/java/org/apache/nifi/processors/aws/kinesis/KinesisConsumerClientTest.java
+++ b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/test/java/org/apache/nifi/processors/aws/kinesis/KinesisConsumerClientTest.java
@@ -204,6 +204,35 @@ class KinesisConsumerClientTest {
verify(mockShardManager, times(2)).readCheckpoint("shardId-000000000001");
}
+ /**
+ * Verifies that polling a result from the per-shard queue resumes a paused EFO consumer.
+ * Without this, single-threaded environments (such as Stateless) starve: the consumer
+ * pauses after delivering one event, and no other thread calls startFetches or
+ * acknowledgeResults to resume it during the consumeRecords loop.
+ */
+ @Test
+ void testPollingResultResumesPausedEfoConsumer() throws Exception {
+ final KinesisShardManager mockShardManager = mock(KinesisShardManager.class);
+ when(mockShardManager.readCheckpoint("shardId-000000000001")).thenReturn("50000");
+
+ final List<SubscribeToShardRequest> capturedRequests = new ArrayList<>();
+ final EnhancedFanOutClient client = createEfoClient(capturedRequests);
+ final List<Shard> shards = List.of(Shard.builder().shardId("shardId-000000000001").build());
+ client.startFetches(shards, "test-stream", 100, "TRIM_HORIZON", mockShardManager);
+
+ final EnhancedFanOutClient.ShardConsumer consumer = client.getShardConsumer("shardId-000000000001");
+ final Subscription subscription = mock(Subscription.class);
+ consumer.setSubscription(subscription);
+ consumer.pause();
+
+ client.enqueueResult(shardFetchResult("shardId-000000000001", "60000"));
+
+ final ShardFetchResult polled = client.pollShardResult("shardId-000000000001");
+ assertNotNull(polled, "Expected queued result to be polled");
+
+ verify(subscription, times(1)).request(1);
+ }
+
/**
* Verifies that acknowledging multiple fetched results from the same shard requests only one
* additional EFO event for that shard.
diff --git a/nifi-stateless/nifi-stateless-api/src/main/java/org/apache/nifi/stateless/config/PropertiesFileEngineConfigurationParser.java b/nifi-stateless/nifi-stateless-api/src/main/java/org/apache/nifi/stateless/config/PropertiesFileEngineConfigurationParser.java
index 502ddfb7946..acf8a251919 100644
--- a/nifi-stateless/nifi-stateless-api/src/main/java/org/apache/nifi/stateless/config/PropertiesFileEngineConfigurationParser.java
+++ b/nifi-stateless/nifi-stateless-api/src/main/java/org/apache/nifi/stateless/config/PropertiesFileEngineConfigurationParser.java
@@ -114,7 +114,7 @@ public class PropertiesFileEngineConfigurationParser {
final String statusTaskInterval = properties.getProperty(STATUS_TASK_INTERVAL, "1 min");
- final String processorStartTimeout = properties.getProperty(PROCESSOR_START_TIMEOUT, "10 secs");
+ final String processorStartTimeout = properties.getProperty(PROCESSOR_START_TIMEOUT, "1 min");
final String componentEnableTimeout = properties.getProperty(COMPONENT_ENABLE_TIMEOUT, "10 secs");
return new StatelessEngineConfiguration() {
diff --git a/nifi-stateless/nifi-stateless-api/src/main/java/org/apache/nifi/stateless/engine/StatelessEngineConfiguration.java b/nifi-stateless/nifi-stateless-api/src/main/java/org/apache/nifi/stateless/engine/StatelessEngineConfiguration.java
index 9df616fe4ea..4b3944dd310 100644
--- a/nifi-stateless/nifi-stateless-api/src/main/java/org/apache/nifi/stateless/engine/StatelessEngineConfiguration.java
+++ b/nifi-stateless/nifi-stateless-api/src/main/java/org/apache/nifi/stateless/engine/StatelessEngineConfiguration.java
@@ -95,7 +95,7 @@ public interface StatelessEngineConfiguration {
* Defaults to "10 secs"
*/
default String getProcessorStartTimeout() {
- return "10 secs";
+ return "1 min";
}
/**