Merge branch 'main' into NIFI-15901

Author: adwk67

nifi-extension-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunk.java

@@ -101,18 +101,26 @@ public void cleanup() {
     }
 
     @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
         final ProcessSession session = sessionFactory.createSession();
         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;
         }
 

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.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 void testCompletingPreviousBatchOnNextExecution() throws Exception {
         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() {

nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/connectable/ConnectionUtils.java

@@ -101,7 +101,8 @@ public static FlowFileCloneResult clone(final FlowFileRecord flowFile, final Col
     }
 
     private static RepositoryRecord createRepositoryRecord(final FlowFileRecord flowFile, final FlowFileQueue destinationQueue) {
-        final StandardRepositoryRecord repoRecord = new StandardRepositoryRecord(null, flowFile);
+        // The FlowFile is being introduced to this repository for the first time, so it is tracked as a CREATE record.
+        final StandardRepositoryRecord repoRecord = new StandardRepositoryRecord(destinationQueue);
         repoRecord.setWorking(flowFile, Collections.emptyMap(), false);
         repoRecord.setDestination(destinationQueue);
         return repoRecord;

nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/connectable/TestConnectionUtils.java

@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.connectable;
+
+import org.apache.nifi.connectable.ConnectionUtils.FlowFileCloneResult;
+import org.apache.nifi.controller.queue.FlowFileQueue;
+import org.apache.nifi.controller.repository.ContentRepository;
+import org.apache.nifi.controller.repository.FlowFileRecord;
+import org.apache.nifi.controller.repository.FlowFileRepository;
+import org.apache.nifi.controller.repository.RepositoryRecord;
+import org.apache.nifi.controller.repository.RepositoryRecordType;
+import org.apache.nifi.controller.repository.StandardFlowFileRecord;
+import org.apache.nifi.controller.repository.claim.ContentClaim;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.when;
+
+public class TestConnectionUtils {
+
+    private FlowFileRepository flowFileRepository;
+    private ContentRepository contentRepository;
+
+    @BeforeEach
+    public void setup() {
+        flowFileRepository = Mockito.mock(FlowFileRepository.class);
+        contentRepository = Mockito.mock(ContentRepository.class);
+
+        final AtomicLong nextId = new AtomicLong(1000L);
+        when(flowFileRepository.getNextFlowFileSequence()).thenAnswer(invocation -> nextId.getAndIncrement());
+    }
+
+    @Test
+    public void testCloneSingleDestinationProducesCreateRecord() {
+        final FlowFileRecord flowFile = new StandardFlowFileRecord.Builder()
+                .id(1L)
+                .addAttribute("uuid", "test-uuid-1")
+                .build();
+
+        final FlowFileQueue destinationQueue = Mockito.mock(FlowFileQueue.class);
+        final Connection destination = mockConnection(destinationQueue);
+
+        final FlowFileCloneResult result = ConnectionUtils.clone(flowFile, List.of(destination), flowFileRepository, contentRepository);
+
+        final List<RepositoryRecord> records = result.getRepositoryRecords();
+        assertEquals(1, records.size());
+        final RepositoryRecord record = records.get(0);
+        assertEquals(RepositoryRecordType.CREATE, record.getType());
+        assertEquals(destinationQueue, record.getDestination());
+        assertNotNull(record.getCurrent());
+        assertEquals(flowFile.getId(), record.getCurrent().getId());
+    }
+
+    @Test
+    public void testCloneMultipleDestinationsAllProduceCreateRecords() {
+        final ContentClaim contentClaim = Mockito.mock(ContentClaim.class);
+        final FlowFileRecord flowFile = new StandardFlowFileRecord.Builder()
+                .id(1L)
+                .addAttribute("uuid", "test-uuid-1")
+                .contentClaim(contentClaim)
+                .size(1024L)
+                .build();
+
+        final FlowFileQueue queueOne = Mockito.mock(FlowFileQueue.class);
+        final FlowFileQueue queueTwo = Mockito.mock(FlowFileQueue.class);
+        final FlowFileQueue queueThree = Mockito.mock(FlowFileQueue.class);
+        final List<Connection> destinations = List.of(
+                mockConnection(queueOne),
+                mockConnection(queueTwo),
+                mockConnection(queueThree));
+
+        final FlowFileCloneResult result = ConnectionUtils.clone(flowFile, destinations, flowFileRepository, contentRepository);
+
+        final List<RepositoryRecord> records = result.getRepositoryRecords();
+        assertEquals(destinations.size(), records.size());
+
+        // Each record produced by clone() represents a FlowFile being introduced to the repository,
+        // whether it is the original routed FlowFile or a sibling clone.
+        assertTrue(records.stream().allMatch(record -> record.getType() == RepositoryRecordType.CREATE));
+
+        // The clones (one per additional destination beyond the first) should each have triggered a
+        // claimant count increment on the shared Content Claim.
+        Mockito.verify(contentRepository, Mockito.times(destinations.size() - 1)).incrementClaimaintCount(contentClaim);
+
+        final List<FlowFileQueue> destinationQueues = new ArrayList<>();
+        for (final RepositoryRecord record : records) {
+            destinationQueues.add(record.getDestination());
+        }
+        assertTrue(destinationQueues.contains(queueOne));
+        assertTrue(destinationQueues.contains(queueTwo));
+        assertTrue(destinationQueues.contains(queueThree));
+    }
+
+    private Connection mockConnection(final FlowFileQueue queue) {
+        final Connection connection = Mockito.mock(Connection.class);
+        when(connection.getFlowFileQueue()).thenReturn(queue);
+        return connection;
+    }
+}