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 <[email protected]>

Author: markap14

nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/ConsumeKinesis.java

@@ -708,7 +708,7 @@ private List<ShardFetchResult> consumeRecords(final Set<String> claimedShards) {
         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()) {

nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/EnhancedFanOutClient.java

@@ -146,6 +146,11 @@ long drainDeduplicatedEventCount() {
         return total;
     }
 
+    @Override
+    protected void onResultPolled() {
+        resumePausedConsumers();
+    }
+
     @Override
     void acknowledgeResults(final List<ShardFetchResult> results) {
         resumePausedConsumers();
@@ -355,6 +360,7 @@ static final class ShardConsumer {
         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;

nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/test/java/org/apache/nifi/processors/aws/kinesis/KinesisConsumerClientTest.java

@@ -204,6 +204,35 @@ void testSubscriptionRenewalAfterPollBeforeAcknowledgeUsesMaxSequence() throws E
         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.

nifi-stateless/nifi-stateless-api/src/main/java/org/apache/nifi/stateless/config/PropertiesFileEngineConfigurationParser.java

@@ -114,7 +114,7 @@ public StatelessEngineConfiguration parseEngineConfiguration(final File properti
 
         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() {

nifi-stateless/nifi-stateless-api/src/main/java/org/apache/nifi/stateless/engine/StatelessEngineConfiguration.java

@@ -95,7 +95,7 @@ default boolean isLogExtensionDiscovery() {
      * Defaults to "10 secs"
      */
     default String getProcessorStartTimeout() {
-        return "10 secs";
+        return "1 min";
     }
 
     /**