From 9eccf0b3932a63c4dc98fcc3d539af21f8053d54 Mon Sep 17 00:00:00 2001 From: Noah Cover Date: Thu, 6 Aug 2026 15:27:34 -0700 Subject: [PATCH 1/3] NIFI-16174 - Treat a stateless process group as a single lifecycle unit when starting/stopping a controller service's referencing components StandardControllerServiceProvider scheduled processors that reference a controller service individually, even when they belong to a stateless process group. Because a stateless group is a single scheduling unit, this left the group with a mixed running/stopped processor state and a group node stuck RUNNING, from which it could not recover. Resolve each referenced processor's owning stateless group (STATELESS -> self, INHERITED -> nearest explicit ancestor) and, for stateless members, stop the group once via ProcessGroup.stopProcessing() / start it once via ComponentScheduler.startStatelessGroup(), mapping the group's single future to every affected member. Standard processors are unchanged. Public ControllerServiceProvider signatures are unchanged. Adds unit coverage in StandardControllerServiceProviderTest and an end-to-end regression (ConnectorTroubleshootingIT) backed by a stateless controller-service reference in the ComponentLifecycleConnector test fixture. --- .../StandardControllerServiceProvider.java | 76 +++++++-- .../service/ControllerServiceProvider.java | 15 ++ ...StandardControllerServiceProviderTest.java | 151 ++++++++++++++++++ .../system/ComponentLifecycleConnector.java | 20 ++- .../ConnectorTroubleshootingIT.java | 107 +++++++++++++ 5 files changed, 351 insertions(+), 18 deletions(-) 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..56186e5117c1 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,24 @@ 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(); + 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..be90de710283 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,21 @@ 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. + // This mirrors real connectors (e.g. a Snowflake connection pool referenced from a "Create Journal Table" + // stateless subgroup) and lets tests exercise stopping/starting the stateless group as a single unit through + // the controller-service reference lifecycle. + 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 From bf3e5cd5487c9542cda03cecef5fffebf84db196 Mon Sep 17 00:00:00 2001 From: Noah Date: Thu, 6 Aug 2026 17:45:40 -0700 Subject: [PATCH 2/3] Update ComponentLifecycleConnector.java --- .../connectors/tests/system/ComponentLifecycleConnector.java | 3 --- 1 file changed, 3 deletions(-) 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 be90de710283..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 @@ -158,9 +158,6 @@ private VersionedProcessGroup createStatelessGroup(final String parentGroupId, f 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. - // This mirrors real connectors (e.g. a Snowflake connection pool referenced from a "Create Journal Table" - // stateless subgroup) and lets tests exercise stopping/starting the stateless group as a single unit through - // the controller-service reference lifecycle. 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); From fc58956410d9d435029ac9aa267493e2b10e41a7 Mon Sep 17 00:00:00 2001 From: Noah Cover Date: Fri, 7 Aug 2026 10:25:20 -0700 Subject: [PATCH 3/3] NIFI-16174 - Guard against null execution engine when resolving the owning stateless group A referenced processor's process group can report a null execution engine (e.g. in unit-test fixtures backed by mock process groups). Treat a null engine as non-stateless so getStatelessGroup returns null and the processor is handled on the standard per-component path, rather than throwing an NPE in the switch. --- .../controller/service/StandardControllerServiceProvider.java | 4 ++++ 1 file changed, 4 insertions(+) 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 56186e5117c1..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 @@ -307,6 +307,10 @@ private ProcessGroup getStatelessGroup(final ProcessGroup start) { } final ExecutionEngine executionEngine = start.getExecutionEngine(); + if (executionEngine == null) { + return null; + } + return switch (executionEngine) { case STATELESS -> start; case INHERITED -> getStatelessGroup(start.getParent());