(nifi) branch main updated: NIFI-15912 - PutSplunk Processor runs countinously and causes high load

[turcsanyi]

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

asf-gitbox-commits 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 873d08e4ab5 NIFI-15912 - PutSplunk Processor runs countinously and 
causes high load
873d08e4ab5 is described below

commit 873d08e4ab5f9c89e6168ab087e7679534eb2a18
Author: Pierre Villard <pierre.villard...@gmail.com>
AuthorDate: Wed May 6 17:30:08 2026 +0200

    NIFI-15912 - PutSplunk Processor runs countinously and causes high load
    
    This closes #11214.
    
    Signed-off-by: Peter Turcsanyi <turcsa...@apache.org>
---
 .../apache/nifi/processors/splunk/PutSplunk.java   | 10 +++++++-
 .../nifi/processors/splunk/TestPutSplunk.java      | 27 ++++++++++++++++++++++
 2 files changed, 36 insertions(+), 1 deletion(-)

diff --git 
a/nifi-extension-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunk.java
 
b/nifi-extension-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunk.java
index 44cea19a79a..74d3cd937bc 100644
--- 
a/nifi-extension-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunk.java
+++ 
b/nifi-extension-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunk.java
@@ -101,11 +101,13 @@ public class PutSplunk extends 
AbstractPutEventProcessor<byte[]> {
     }
 
     @Override
-    public void onTrigger(ProcessContext context, ProcessSessionFactory 
sessionFactory) throws ProcessException {
+    public void onTrigger(final ProcessContext context, final 
ProcessSessionFactory sessionFactory) throws ProcessException {
         // first complete any batches from previous executions
+        boolean completedAny = false;
         FlowFileMessageBatch batch;
         while ((batch = completeBatches.poll()) != null) {
             batch.completeSession();
+            completedAny = true;
         }
 
         // create a session and try to get a FlowFile, if none available then 
close any idle senders
@@ -113,6 +115,12 @@ public class PutSplunk extends 
AbstractPutEventProcessor<byte[]> {
         final FlowFile flowFile = session.get();
 
         if (flowFile == null) {
+            // The processor is annotated with @TriggerWhenEmpty so onTrigger 
is invoked even with no input,
+            // allowing async send callbacks to drain completeBatches. Yield 
when nothing was drained to avoid
+            // a busy scheduling loop on an idle processor.
+            if (!completedAny) {
+                context.yield();
+            }
             return;
         }
 
diff --git 
a/nifi-extension-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/test/java/org/apache/nifi/processors/splunk/TestPutSplunk.java
 
b/nifi-extension-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/test/java/org/apache/nifi/processors/splunk/TestPutSplunk.java
index 9d8c25a8043..7f9f29b0c6b 100644
--- 
a/nifi-extension-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/test/java/org/apache/nifi/processors/splunk/TestPutSplunk.java
+++ 
b/nifi-extension-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/test/java/org/apache/nifi/processors/splunk/TestPutSplunk.java
@@ -42,8 +42,10 @@ import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class TestPutSplunk {
 
@@ -261,6 +263,31 @@ public class TestPutSplunk {
         checkReceivedAllData(message);
     }
 
+    @Test
+    @Timeout(value = DEFAULT_TEST_TIMEOUT_PERIOD, unit = TimeUnit.MILLISECONDS)
+    public void testYieldsWhenIdle() throws Exception {
+        createTestServer(TransportProtocol.TCP);
+
+        runner.run(1);
+
+        assertTrue(runner.isYieldCalled(), "Processor should yield when no 
FlowFile is available to avoid busy scheduling");
+    }
+
+    @Test
+    @Timeout(value = DEFAULT_TEST_TIMEOUT_PERIOD, unit = TimeUnit.MILLISECONDS)
+    public void testDoesNotYieldWhenFlowFileProcessed() throws Exception {
+        createTestServer(TransportProtocol.TCP);
+        final String message = "This is one message, should send the whole 
FlowFile";
+
+        runner.enqueue(message);
+        runner.run(1);
+        runner.assertAllFlowFilesTransferred(PutSplunk.REL_SUCCESS, 1);
+
+        checkReceivedAllData(message);
+
+        assertFalse(runner.isYieldCalled(), "Processor should not yield after 
successfully processing a FlowFile");
+    }
+
     @Test
     @Timeout(value = DEFAULT_TEST_TIMEOUT_PERIOD, unit = TimeUnit.MILLISECONDS)
     public void testUnableToCreateConnectionShouldRouteToFailure() {