NIFI-16174 - Treat a stateless process group as a single lifecycle unit when starting/stopping a controller service's referencing components

nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java

@@ -27,6 +27,7 @@
 import org.apache.nifi.controller.ScheduledState;
 import org.apache.nifi.controller.flow.FlowManager;
 import org.apache.nifi.events.BulletinFactory;
+import org.apache.nifi.flow.ExecutionEngine;
 import org.apache.nifi.groups.ComponentScheduler;
 import org.apache.nifi.groups.DefaultComponentScheduler;
 import org.apache.nifi.groups.ProcessGroup;
@@ -148,17 +149,27 @@ public Set<ComponentNode> scheduleReferencingComponents(final ControllerServiceN
             }
         }
 
-        // start all of the components that are not disabled
+        // Start each component that is not disabled. A processor in a stateless group is started through its group
+        // (the group is a single scheduling unit), and each stateless group is started at most once.
         final Set<ComponentNode> updated = new HashSet<>();
+        final Set<ProcessGroup> startedStatelessGroups = new HashSet<>();
         for (final ProcessorNode node : processors) {
             if (candidates != null && !candidates.contains(node)) {
                 continue;
             }
 
-            if (node.getScheduledState() != ScheduledState.DISABLED) {
+            if (node.getScheduledState() == ScheduledState.DISABLED) {
+                continue;
+            }
+
+            final ProcessGroup statelessGroup = getStatelessGroup(node.getProcessGroup());
+            if (statelessGroup == null) {
                 componentScheduler.startComponent(node);
-                updated.add(node);
+            } else if (startedStatelessGroups.add(statelessGroup)) {
+                componentScheduler.startStatelessGroup(statelessGroup);
             }
+
+            updated.add(node);
         }
         for (final ReportingTaskNode node : reportingTasks) {
             if (candidates != null && !candidates.contains(node)) {
@@ -194,15 +205,30 @@ public Map<ComponentNode, Future<Void>> unscheduleReferencingComponents(final Co
 
         final Map<ComponentNode, Future<Void>> updated = new HashMap<>();
 
+        // Partition running/starting processors: those in a stateless group are stopped through the group (a single
+        // scheduling unit); standard processors are stopped individually.
+        final Map<ProcessGroup, List<ProcessorNode>> statelessMembersByGroup = new HashMap<>();
+        final List<ProcessorNode> standardProcessors = new ArrayList<>();
+        for (final ProcessorNode node : processors) {
+            if (!isRunningOrStarting(node)) {
+                continue;
+            }
+
+            final ProcessGroup statelessGroup = getStatelessGroup(node.getProcessGroup());
+            if (statelessGroup == null) {
+                standardProcessors.add(node);
+            } else {
+                statelessMembersByGroup.computeIfAbsent(statelessGroup, group -> new ArrayList<>()).add(node);
+            }
+        }
+
         // verify that we can stop all components (that are running or starting) before doing anything
         // Note: We check both RUNNING and STARTING states because a processor might be stuck in STARTING
         // state if it references an invalid controller service (e.g., after a restart when the controller
         // service configuration became invalid). Such processors need to be stopped before the controller
-        // service can be disabled.
-        for (final ProcessorNode node : processors) {
-            if (isRunningOrStarting(node)) {
-                node.verifyCanStop();
-            }
+        // service can be disabled. Stateless-group members are verified and stopped through their group.
+        for (final ProcessorNode node : standardProcessors) {
+            node.verifyCanStop();
         }
         for (final ReportingTaskNode node : reportingTasks) {
             if (isRunningOrStarting(node)) {
@@ -215,13 +241,19 @@ public Map<ComponentNode, Future<Void>> unscheduleReferencingComponents(final Co
             }
         }
 
-        // stop all of the components that are running or starting
-        for (final ProcessorNode node : processors) {
-            if (isRunningOrStarting(node)) {
-                final Future<Void> future = node.getProcessGroup().stopProcessor(node);
-                updated.put(node, future);
+        // stop each stateless group once as a single unit, mapping the group's single future to every affected member
+        for (final Map.Entry<ProcessGroup, List<ProcessorNode>> entry : statelessMembersByGroup.entrySet()) {
+            final Future<Void> future = entry.getKey().stopProcessing();
+            for (final ProcessorNode member : entry.getValue()) {
+                updated.put(member, future);
             }
         }
+
+        // stop the standard processors
+        for (final ProcessorNode node : standardProcessors) {
+            final Future<Void> future = node.getProcessGroup().stopProcessor(node);
+            updated.put(node, future);
+        }
         for (final ReportingTaskNode node : reportingTasks) {
             if (isRunningOrStarting(node)) {
                 final Future<Void> future = processScheduler.unschedule(node);
@@ -264,6 +296,28 @@ private boolean isRunningOrStarting(final ReportingTaskNode node) {
         return scheduledState == ScheduledState.RUNNING || scheduledState == ScheduledState.STARTING;
     }
 
+    /**
+     * Returns the explicit stateless process group that the given process group belongs to, or {@code null} if the
+     * process group is not part of a stateless group. A group whose execution engine is {@code INHERITED} resolves to
+     * its nearest ancestor that explicitly declares an execution engine.
+     */
+    private ProcessGroup getStatelessGroup(final ProcessGroup start) {
+        if (start == null) {
+            return null;
+        }
+
+        final ExecutionEngine executionEngine = start.getExecutionEngine();
+        if (executionEngine == null) {
+            return null;
+        }
+
+        return switch (executionEngine) {
+            case STATELESS -> start;
+            case INHERITED -> getStatelessGroup(start.getParent());
+            default -> null;
+        };
+    }
+
     @Override
     public CompletableFuture<Void> enableControllerService(final ControllerServiceNode serviceNode) {
         if (serviceNode.isActive()) {

nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java

@@ -134,7 +134,15 @@ public interface ControllerServiceProvider extends ControllerServiceLookup {
      * Controller services that reference this one, its schedulable referencing
      * components will also be unscheduled.
      *
+     * A referencing processor that belongs to a stateless process group is not
+     * stopped individually; instead its owning stateless group is stopped as a
+     * single unit, and the group's single stop {@link Future} is mapped to every
+     * affected processor in that group in the returned map.
+     *
      * @param serviceNode the node
+     *
+     * @return a map of each affected component to the {@link Future} that
+     * completes when the component (or its stateless group) has stopped
      */
     Map<ComponentNode, Future<Void>> unscheduleReferencingComponents(ControllerServiceNode serviceNode);
 
@@ -199,12 +207,19 @@ public interface ControllerServiceProvider extends ControllerServiceLookup {
      * recursively, so if a Processor is referencing Service A, which is
      * referencing serviceNode, then the Processor will also be started.
      *
+     * A referencing processor that belongs to a stateless process group is not
+     * started individually; instead its owning stateless group is started as a
+     * single unit. Each affected stateless group is started at most once even
+     * when several of its processors reference the service.
+     *
      * @param serviceNode the node
      */
     Set<ComponentNode> scheduleReferencingComponents(ControllerServiceNode serviceNode);
 
     /**
      * Schedules any of the candidate components that are currently referencing the given Controller Service to run.
+     * A candidate processor that belongs to a stateless process group causes its owning stateless group to be started
+     * as a single unit; the group is started once even if only one of its processors is among the candidates.
      * @return the components that were scheduled
      */
     Set<ComponentNode> scheduleReferencingComponents(ControllerServiceNode serviceNode, Set<ComponentNode> candidates, ComponentScheduler componentScheduler);

nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java

@@ -21,12 +21,19 @@
 import org.apache.nifi.components.state.StateManagerProvider;
 import org.apache.nifi.components.validation.ValidationTrigger;
 import org.apache.nifi.components.validation.VerifiableComponentFactory;
+import org.apache.nifi.controller.ComponentNode;
 import org.apache.nifi.controller.ControllerService;
 import org.apache.nifi.controller.ExtensionBuilder;
+import org.apache.nifi.controller.FlowAnalysisRuleNode;
 import org.apache.nifi.controller.NodeTypeProvider;
 import org.apache.nifi.controller.ProcessScheduler;
+import org.apache.nifi.controller.ProcessorNode;
 import org.apache.nifi.controller.ReloadComponent;
+import org.apache.nifi.controller.ReportingTaskNode;
+import org.apache.nifi.controller.ScheduledState;
 import org.apache.nifi.controller.flow.FlowManager;
+import org.apache.nifi.flow.ExecutionEngine;
+import org.apache.nifi.groups.ComponentScheduler;
 import org.apache.nifi.groups.ProcessGroup;
 import org.apache.nifi.nar.ExtensionDiscoveringManager;
 import org.apache.nifi.nar.ExtensionManager;
@@ -50,14 +57,17 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -236,6 +246,147 @@ void testGetControllerServiceIdentifiersForConnectorManagedGroup() {
         assertTrue(identifiers.contains(serviceId));
     }
 
+    @Test
+    void testUnscheduleReferencingComponentsStopsStatelessGroupAsUnit() {
+        final ProcessGroup statelessGroup = createGroup(ExecutionEngine.STATELESS, null);
+        final ProcessGroup standardGroup = createGroup(ExecutionEngine.STANDARD, null);
+
+        final ProcessorNode statelessProcessorA = createProcessor(statelessGroup, ScheduledState.RUNNING, ScheduledState.RUNNING);
+        final ProcessorNode statelessProcessorB = createProcessor(statelessGroup, ScheduledState.RUNNING, ScheduledState.RUNNING);
+        final ProcessorNode standardProcessor = createProcessor(standardGroup, ScheduledState.RUNNING, ScheduledState.RUNNING);
+
+        when(statelessGroup.stopProcessing()).thenReturn(CompletableFuture.completedFuture(null));
+        when(standardGroup.stopProcessor(standardProcessor)).thenReturn(CompletableFuture.completedFuture(null));
+
+        final ControllerServiceNode serviceNode = createServiceWithProcessorReferences(List.of(statelessProcessorA, statelessProcessorB, standardProcessor));
+
+        final Map<ComponentNode, Future<Void>> result = serviceProvider.unscheduleReferencingComponents(serviceNode);
+
+        // The stateless group is stopped once as a single unit; its member processors are not stopped individually.
+        verify(statelessGroup, times(1)).stopProcessing();
+        verify(statelessGroup, never()).stopProcessor(any());
+        verify(standardGroup, never()).stopProcessing();
+
+        // The standard processor is stopped directly through its process group.
+        verify(standardGroup, times(1)).stopProcessor(standardProcessor);
+
+        // Every affected processor is represented in the returned map, and the stateless members share the group's single future.
+        assertTrue(result.containsKey(statelessProcessorA));
+        assertTrue(result.containsKey(statelessProcessorB));
+        assertTrue(result.containsKey(standardProcessor));
+        assertEquals(result.get(statelessProcessorA), result.get(statelessProcessorB));
+    }
+
+    @Test
+    void testUnscheduleReferencingComponentsResolvesInheritedChildToStatelessAncestor() {
+        final ProcessGroup statelessGroup = createGroup(ExecutionEngine.STATELESS, null);
+        final ProcessGroup inheritedChild = createGroup(ExecutionEngine.INHERITED, statelessGroup);
+        final ProcessorNode childProcessor = createProcessor(inheritedChild, ScheduledState.RUNNING, ScheduledState.RUNNING);
+
+        when(statelessGroup.stopProcessing()).thenReturn(CompletableFuture.completedFuture(null));
+
+        final ControllerServiceNode serviceNode = createServiceWithProcessorReferences(List.of(childProcessor));
+
+        final Map<ComponentNode, Future<Void>> result = serviceProvider.unscheduleReferencingComponents(serviceNode);
+
+        // A processor in an INHERITED child resolves up to its explicit stateless ancestor, which is stopped as a unit.
+        verify(statelessGroup, times(1)).stopProcessing();
+        verify(inheritedChild, never()).stopProcessing();
+        verify(inheritedChild, never()).stopProcessor(any());
+        assertTrue(result.containsKey(childProcessor));
+    }
+
+    @Test
+    void testScheduleReferencingComponentsStartsStatelessGroupAsUnit() {
+        final ProcessGroup statelessGroup = createGroup(ExecutionEngine.STATELESS, null);
+        final ProcessGroup standardGroup = createGroup(ExecutionEngine.STANDARD, null);
+
+        final ProcessorNode statelessProcessorA = createProcessor(statelessGroup, ScheduledState.STOPPED, ScheduledState.STOPPED);
+        final ProcessorNode statelessProcessorB = createProcessor(statelessGroup, ScheduledState.STOPPED, ScheduledState.STOPPED);
+        final ProcessorNode standardProcessor = createProcessor(standardGroup, ScheduledState.STOPPED, ScheduledState.STOPPED);
+
+        final ControllerServiceNode serviceNode = createServiceWithProcessorReferences(List.of(statelessProcessorA, statelessProcessorB, standardProcessor));
+        final ComponentScheduler componentScheduler = mock(ComponentScheduler.class);
+
+        final Set<ComponentNode> result = serviceProvider.scheduleReferencingComponents(serviceNode, null, componentScheduler);
+
+        // The stateless group is started once as a single unit; its members are not started individually.
+        verify(componentScheduler, times(1)).startStatelessGroup(statelessGroup);
+        verify(componentScheduler, never()).startComponent(statelessProcessorA);
+        verify(componentScheduler, never()).startComponent(statelessProcessorB);
+        verify(componentScheduler, never()).startStatelessGroup(standardGroup);
+
+        // The standard processor is started directly.
+        verify(componentScheduler, times(1)).startComponent(standardProcessor);
+
+        assertTrue(result.contains(statelessProcessorA));
+        assertTrue(result.contains(statelessProcessorB));
+        assertTrue(result.contains(standardProcessor));
+    }
+
+    @Test
+    void testScheduleReferencingComponentsWithCandidateStartsOwningStatelessGroup() {
+        final ProcessGroup statelessGroup = createGroup(ExecutionEngine.STATELESS, null);
+
+        final ProcessorNode statelessProcessorA = createProcessor(statelessGroup, ScheduledState.STOPPED, ScheduledState.STOPPED);
+        final ProcessorNode statelessProcessorB = createProcessor(statelessGroup, ScheduledState.STOPPED, ScheduledState.STOPPED);
+
+        final ControllerServiceNode serviceNode = createServiceWithProcessorReferences(List.of(statelessProcessorA, statelessProcessorB));
+        final ComponentScheduler componentScheduler = mock(ComponentScheduler.class);
+
+        // Only one member of the stateless group is a candidate, but the group must still be started as a whole.
+        final Set<ComponentNode> result = serviceProvider.scheduleReferencingComponents(serviceNode, Set.of(statelessProcessorA), componentScheduler);
+
+        verify(componentScheduler, times(1)).startStatelessGroup(statelessGroup);
+        verify(componentScheduler, never()).startComponent(statelessProcessorA);
+        assertTrue(result.contains(statelessProcessorA));
+    }
+
+    @Test
+    void testUnscheduleReferencingComponentsStopsEachStatelessGroupOnce() {
+        final ProcessGroup firstGroup = createGroup(ExecutionEngine.STATELESS, null);
+        final ProcessGroup secondGroup = createGroup(ExecutionEngine.STATELESS, null);
+
+        final ProcessorNode firstMember = createProcessor(firstGroup, ScheduledState.RUNNING, ScheduledState.RUNNING);
+        final ProcessorNode secondMember = createProcessor(secondGroup, ScheduledState.RUNNING, ScheduledState.RUNNING);
+
+        when(firstGroup.stopProcessing()).thenReturn(CompletableFuture.completedFuture(null));
+        when(secondGroup.stopProcessing()).thenReturn(CompletableFuture.completedFuture(null));
+
+        final ControllerServiceNode serviceNode = createServiceWithProcessorReferences(List.of(firstMember, secondMember));
+
+        serviceProvider.unscheduleReferencingComponents(serviceNode);
+
+        // Two distinct stateless groups are each stopped exactly once.
+        verify(firstGroup, times(1)).stopProcessing();
+        verify(secondGroup, times(1)).stopProcessing();
+    }
+
+    private ProcessGroup createGroup(final ExecutionEngine executionEngine, final ProcessGroup parent) {
+        final ProcessGroup group = mock(ProcessGroup.class);
+        lenient().when(group.getExecutionEngine()).thenReturn(executionEngine);
+        lenient().when(group.getParent()).thenReturn(parent);
+        return group;
+    }
+
+    private ProcessorNode createProcessor(final ProcessGroup group, final ScheduledState scheduledState, final ScheduledState physicalState) {
+        final ProcessorNode processor = mock(ProcessorNode.class);
+        lenient().when(processor.getProcessGroup()).thenReturn(group);
+        lenient().when(processor.getScheduledState()).thenReturn(scheduledState);
+        lenient().when(processor.getPhysicalScheduledState()).thenReturn(physicalState);
+        return processor;
+    }
+
+    private ControllerServiceNode createServiceWithProcessorReferences(final List<ProcessorNode> processors) {
+        final ControllerServiceNode serviceNode = mock(ControllerServiceNode.class);
+        final ControllerServiceReference reference = mock(ControllerServiceReference.class);
+        when(serviceNode.getReferences()).thenReturn(reference);
+        when(reference.findRecursiveReferences(ProcessorNode.class)).thenReturn(processors);
+        when(reference.findRecursiveReferences(ReportingTaskNode.class)).thenReturn(Collections.emptyList());
+        when(reference.findRecursiveReferences(FlowAnalysisRuleNode.class)).thenReturn(Collections.emptyList());
+        return serviceNode;
+    }
+
     private ControllerServiceNode createControllerService(
             final String identifier,
             final String type,