diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java index 4e0d5dadf433..81893f67921a 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java +++ b/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 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 updated = new HashSet<>(); + final Set 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> unscheduleReferencingComponents(final Co final Map> 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> statelessMembersByGroup = new HashMap<>(); + final List 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> unscheduleReferencingComponents(final Co } } - // stop all of the components that are running or starting - for (final ProcessorNode node : processors) { - if (isRunningOrStarting(node)) { - final Future 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> entry : statelessMembersByGroup.entrySet()) { + final Future 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 future = node.getProcessGroup().stopProcessor(node); + updated.put(node, future); + } for (final ReportingTaskNode node : reportingTasks) { if (isRunningOrStarting(node)) { final Future 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 enableControllerService(final ControllerServiceNode serviceNode) { if (serviceNode.isActive()) { diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java index 6e3755427964..362e1b814be8 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java +++ b/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> 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 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 scheduleReferencingComponents(ControllerServiceNode serviceNode, Set candidates, ComponentScheduler componentScheduler); diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java index 36be3f53163a..54d7b2b4ab65 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java +++ b/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,6 +57,7 @@ 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; @@ -57,7 +65,9 @@ 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> 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> 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 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 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 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, diff --git a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/connectors/tests/system/ComponentLifecycleConnector.java b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/connectors/tests/system/ComponentLifecycleConnector.java index 9eddbf597e02..6fcd4b757dee 100644 --- a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/connectors/tests/system/ComponentLifecycleConnector.java +++ b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/connectors/tests/system/ComponentLifecycleConnector.java @@ -95,7 +95,7 @@ private VersionedProcessGroup createRootGroup() { final VersionedProcessor rootTerminateProcessor = VersionedFlowUtils.addProcessor(rootGroup, "org.apache.nifi.processors.tests.system.TerminateFlowFile", SYSTEM_TEST_EXTENSIONS_BUNDLE, "Root TerminateFlowFile", new Position(300, 100)); - final VersionedProcessGroup childGroup = createChildGroup(rootGroup.getIdentifier()); + final VersionedProcessGroup childGroup = createChildGroup(rootGroup.getIdentifier(), rootControllerService.getIdentifier()); rootGroup.getProcessGroups().add(childGroup); final VersionedPort childInputPort = childGroup.getInputPorts().iterator().next(); @@ -109,7 +109,7 @@ private VersionedProcessGroup createRootGroup() { return rootGroup; } - private VersionedProcessGroup createChildGroup(final String parentGroupId) { + private VersionedProcessGroup createChildGroup(final String parentGroupId, final String rootCountServiceId) { final VersionedProcessGroup childGroup = VersionedFlowUtils.createProcessGroup("child-group-id", "Child Group"); childGroup.setPosition(new Position(100, 300)); childGroup.setRemoteProcessGroups(new HashSet<>()); @@ -131,7 +131,7 @@ private VersionedProcessGroup createChildGroup(final String parentGroupId) { final VersionedProcessor childProcessor = VersionedFlowUtils.addProcessor(childGroup, "org.apache.nifi.processors.tests.system.PassThrough", SYSTEM_TEST_EXTENSIONS_BUNDLE, "Child Terminate", new Position(100, 100)); - final VersionedProcessGroup statelessGroup = createStatelessGroup(childGroup.getIdentifier()); + final VersionedProcessGroup statelessGroup = createStatelessGroup(childGroup.getIdentifier(), rootCountServiceId); childGroup.getProcessGroups().add(statelessGroup); final VersionedPort statelessInputPort = statelessGroup.getInputPorts().iterator().next(); @@ -146,7 +146,7 @@ private VersionedProcessGroup createChildGroup(final String parentGroupId) { return childGroup; } - private VersionedProcessGroup createStatelessGroup(final String parentGroupId) { + private VersionedProcessGroup createStatelessGroup(final String parentGroupId, final String rootCountServiceId) { final VersionedProcessGroup statelessGroup = VersionedFlowUtils.createProcessGroup("stateless-group-id", "Stateless Group"); statelessGroup.setPosition(new Position(400, 100)); statelessGroup.setRemoteProcessGroups(new HashSet<>()); @@ -157,11 +157,18 @@ private VersionedProcessGroup createStatelessGroup(final String parentGroupId) { final VersionedPort statelessInput = VersionedFlowUtils.addInputPort(statelessGroup, "Stateless Input", new Position(0, 0)); + // A processor inside the stateless group that references a controller service defined at the connector root. + final VersionedProcessor statelessCountProcessor = VersionedFlowUtils.addProcessor(statelessGroup, + "org.apache.nifi.processors.tests.system.CountFlowFiles", SYSTEM_TEST_EXTENSIONS_BUNDLE, "Stateless Count", new Position(100, 50)); + statelessCountProcessor.getProperties().put("Count Service", rootCountServiceId); + final VersionedProcessor statelessProcessor = VersionedFlowUtils.addProcessor(statelessGroup, "org.apache.nifi.processors.tests.system.TerminateFlowFile", SYSTEM_TEST_EXTENSIONS_BUNDLE, "Stateless Terminate", new Position(100, 100)); VersionedFlowUtils.addConnection(statelessGroup, VersionedFlowUtils.createConnectableComponent(statelessInput), - VersionedFlowUtils.createConnectableComponent(statelessProcessor), Set.of("")); + VersionedFlowUtils.createConnectableComponent(statelessCountProcessor), Set.of("")); + VersionedFlowUtils.addConnection(statelessGroup, VersionedFlowUtils.createConnectableComponent(statelessCountProcessor), + VersionedFlowUtils.createConnectableComponent(statelessProcessor), Set.of("success")); return statelessGroup; } diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorTroubleshootingIT.java b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorTroubleshootingIT.java index 1bbe86c3f21b..04f753529c8d 100644 --- a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorTroubleshootingIT.java +++ b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorTroubleshootingIT.java @@ -36,6 +36,8 @@ import org.apache.nifi.web.api.entity.ConnectionEntity; import org.apache.nifi.web.api.entity.ConnectorEntity; import org.apache.nifi.web.api.entity.ControllerServiceEntity; +import org.apache.nifi.web.api.entity.ControllerServiceReferencingComponentEntity; +import org.apache.nifi.web.api.entity.ControllerServiceReferencingComponentsEntity; import org.apache.nifi.web.api.entity.HistoryEntity; import org.apache.nifi.web.api.entity.ParameterProviderEntity; import org.apache.nifi.web.api.entity.PortEntity; @@ -43,6 +45,7 @@ import org.apache.nifi.web.api.entity.ProcessGroupFlowEntity; import org.apache.nifi.web.api.entity.ProcessorEntity; import org.apache.nifi.web.api.entity.ScheduleComponentsEntity; +import org.apache.nifi.web.api.entity.UpdateControllerServiceReferenceRequestEntity; import org.junit.jupiter.api.Test; import java.io.File; @@ -66,6 +69,72 @@ */ public class ConnectorTroubleshootingIT extends NiFiSystemIT { + /** + * Regression test for stateless-group handling in the controller-service reference lifecycle. A processor inside a + * stateless group references a controller service defined at the connector root. Stopping and starting that + * service's referencing components through the controller-service references endpoint must transition the entire + * stateless group as a single unit - including a sibling processor in the group that does not itself reference the + * service - rather than leaving the group in a mixed running/stopped state. + */ + @Test + public void testControllerServiceReferenceLifecycleTransitionsStatelessGroupAsUnit() throws NiFiClientException, IOException, InterruptedException { + final ConnectorEntity connector = getClientUtil().createConnector("ComponentLifecycleConnector"); + final String connectorId = connector.getId(); + + getClientUtil().applyConnectorUpdate(connector); + getClientUtil().waitForValidConnector(connectorId); + + getClientUtil().enterTroubleshooting(connectorId); + assertConnectorState(connectorId, ConnectorState.TROUBLESHOOTING); + + // The stateless group contains two processors; only one of them references the root controller service. + final List statelessProcessors = findStatelessProcessors(connectorId); + assertEquals(2, statelessProcessors.size(), "Stateless group should contain exactly two processors"); + final List statelessProcessorIds = statelessProcessors.stream().map(ProcessorEntity::getId).toList(); + + final String rootServiceId = findFirstControllerServiceId(connectorId); + assertNotNull(rootServiceId, "Managed flow should contain the root controller service"); + + // Enable the root controller service so its referencing components can be scheduled. + enableControllerService(rootServiceId); + + final ControllerServiceReferencingComponentsEntity references = + getNifiClient().getControllerServicesClient().getControllerServiceReferences(rootServiceId); + final List referencingProcessorIds = references.getControllerServiceReferencingComponents().stream() + .map(ControllerServiceReferencingComponentEntity::getId) + .toList(); + assertFalse(referencingProcessorIds.isEmpty(), "Root controller service should have at least one referencing processor"); + assertTrue(statelessProcessorIds.containsAll(referencingProcessorIds), + "Every referencing processor should be inside the stateless group"); + assertFalse(referencingProcessorIds.containsAll(statelessProcessorIds), + "At least one stateless processor should not directly reference the service, to prove group-as-unit behavior"); + + // Start the referencing components; the whole stateless group must start, including the sibling that does not + // reference the service. + updateReferenceState(rootServiceId, references, ScheduledState.RUNNING.name()); + for (final String processorId : statelessProcessorIds) { + waitForProcessorState(processorId, ScheduledState.RUNNING); + } + + // Stop the referencing components; the whole stateless group must stop as a unit, with no mixed state. + final ControllerServiceReferencingComponentsEntity runningReferences = + getNifiClient().getControllerServicesClient().getControllerServiceReferences(rootServiceId); + updateReferenceState(rootServiceId, runningReferences, ScheduledState.STOPPED.name()); + for (final String processorId : statelessProcessorIds) { + waitForProcessorState(processorId, ScheduledState.STOPPED); + } + + // Disable services and exit Troubleshooting, then confirm the Connector starts cleanly with no mixed state. + final String managedGroupId = getNifiClient().getConnectorClient().getConnector(connectorId).getComponent().getManagedProcessGroupId(); + getClientUtil().disableControllerServices(managedGroupId, true); + + getClientUtil().endTroubleshooting(connectorId); + assertConnectorState(connectorId, ConnectorState.STOPPED); + + getClientUtil().startConnector(connectorId); + assertConnectorState(connectorId, ConnectorState.RUNNING); + } + /** * Transition a Connector into Troubleshooting, modify a processor inside the managed flow, then transition back * out. The Connector's authoritative flow should be restored on exit and the Connector should start smoothly. @@ -559,6 +628,44 @@ private void assertConnectorState(final String connectorId, final ConnectorState assertEquals(expected.name(), entity.getComponent().getState()); } + private List findStatelessProcessors(final String connectorId) throws NiFiClientException, IOException { + final List result = new ArrayList<>(); + final Map statelessByGroupId = new HashMap<>(); + for (final ProcessorEntity processor : findAllProcessors(connectorId)) { + final String parentGroupId = processor.getComponent().getParentGroupId(); + Boolean stateless = statelessByGroupId.get(parentGroupId); + if (stateless == null) { + final ProcessGroupEntity parentGroup = getNifiClient().getProcessGroupClient().getProcessGroup(parentGroupId); + stateless = "STATELESS".equals(parentGroup.getComponent().getExecutionEngine()); + statelessByGroupId.put(parentGroupId, stateless); + } + + if (stateless) { + result.add(processor); + } + } + return result; + } + + private void enableControllerService(final String serviceId) throws NiFiClientException, IOException, InterruptedException { + final ControllerServiceEntity service = getNifiClient().getControllerServicesClient().getControllerService(serviceId); + getClientUtil().enableControllerService(service); + getClientUtil().waitForControllerServiceRunStatus(serviceId, "ENABLED"); + } + + private void updateReferenceState(final String serviceId, final ControllerServiceReferencingComponentsEntity references, final String state) throws NiFiClientException, IOException { + final Map revisions = new HashMap<>(); + for (final ControllerServiceReferencingComponentEntity component : references.getControllerServiceReferencingComponents()) { + revisions.put(component.getId(), component.getRevision()); + } + + final UpdateControllerServiceReferenceRequestEntity request = new UpdateControllerServiceReferenceRequestEntity(); + request.setId(serviceId); + request.setReferencingComponentRevisions(revisions); + request.setState(state); + getNifiClient().getControllerServicesClient().updateControllerServiceReferences(request); + } + /** * Stop and restart the NiFi instance, then wait for all nodes to reconnect when running in a clustered * environment. Subsequent flow-modifying requests (such as {@code endTroubleshooting}) would otherwise be rejected