You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2026/08/28 20:15:06 UTC
(nifi) branch main updated: NIFI-16174 - Treat a stateless process group as a single lifecycle unit when starting/stopping a controller service's referencing components (#11515)
This is an automated email from the ASF dual-hosted git repository.
markap14 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 2597636c8e3 NIFI-16174 - Treat a stateless process group as a single lifecycle unit when starting/stopping a controller service's referencing components (#11515)
2597636c8e3 is described below
commit 2597636c8e363725d2cc0769f360bcf77ad8378c
Author: Noah <no...@gmail.com>
AuthorDate: Fri Aug 28 13:15:00 2026 -0700
NIFI-16174 - Treat a stateless process group as a single lifecycle unit when starting/stopping a controller service's referencing components (#11515)
* 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.
* Update ComponentLifecycleConnector.java
* 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.
* NIFI-16174 - Resolve the top-most stateless group when starting/stopping a controller service's referencing components
The previous helper returned the inner-most Process Group marked STATELESS. Only
the top-most stateless group may be started or stopped directly, so on a nested
stateless chain both paths became silent no-ops: startProcessing() logs a warning
and returns, and stopProcessing() returns an already-completed Future without
stopping anything.
Walk up while the group resolves to STATELESS and operate on the last one.
Document the rule on ProcessGroup.startProcessing()/stopProcessing(), and pin the
no-op with a test, since it is what terminates the recursive stopComponents() walk
during stateless shutdown.
The system test now carries a second stateless subtree whose only referencing
processor lives in a nested stateless group, which is what makes it discriminate:
when a referencing processor exists in the outer group too, the outer group's
transition masks the nested no-op.
---
.../service/StandardControllerServiceProvider.java | 79 ++++++--
.../nifi/groups/StandardProcessGroupTest.java | 26 +++
.../service/ControllerServiceProvider.java | 15 ++
.../StandardControllerServiceProviderTest.java | 216 +++++++++++++++++++++
.../tests/system/ComponentLifecycleConnector.java | 81 ++++++--
.../connectors/ConnectorTroubleshootingIT.java | 109 +++++++++++
6 files changed, 496 insertions(+), 30 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 4e0d5dadf43..fe1f98127fe 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.ReportingTaskNode;
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 class StandardControllerServiceProvider implements ControllerServiceProvi
}
}
- // start all of the components that are 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 = getTopMostStatelessGroup(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 class StandardControllerServiceProvider implements ControllerServiceProvi
final Map<ComponentNode, Future<Void>> updated = new HashMap<>();
+ // A processor in a stateless group is 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 = getTopMostStatelessGroup(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 class StandardControllerServiceProvider implements ControllerServiceProvi
}
}
- // 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,27 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
return scheduledState == ScheduledState.RUNNING || scheduledState == ScheduledState.STARTING;
}
+ /**
+ * Returns the top-most Process Group that runs using the Stateless Engine and contains the given Process Group, or
+ * {@code null} if the given Process Group is {@code null} or does not run using the Stateless Engine. Only the
+ * top-most stateless group may be started or stopped directly; {@link ProcessGroup#startProcessing()} and
+ * {@link ProcessGroup#stopProcessing()} are no-ops on a nested stateless group.
+ */
+ private ProcessGroup getTopMostStatelessGroup(final ProcessGroup start) {
+ ProcessGroup topMost = null;
+
+ // a Standard-Engine group cannot be a child of a stateless group, so the stateless chain is contiguous
+ for (ProcessGroup group = start; group != null; group = group.getParent()) {
+ if (group.resolveExecutionEngine() != ExecutionEngine.STATELESS) {
+ break;
+ }
+
+ topMost = group;
+ }
+
+ return topMost;
+ }
+
@Override
public CompletableFuture<Void> enableControllerService(final ControllerServiceNode serviceNode) {
if (serviceNode.isActive()) {
diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java
index 070bd5ad42a..0e46d63531f 100644
--- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java
+++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java
@@ -28,6 +28,7 @@ import org.apache.nifi.controller.ReloadComponent;
import org.apache.nifi.controller.flow.FlowManager;
import org.apache.nifi.controller.service.ControllerServiceProvider;
import org.apache.nifi.encrypt.PropertyEncryptor;
+import org.apache.nifi.flow.ExecutionEngine;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.registry.flow.VersionControlInformation;
import org.apache.nifi.registry.flow.VersionedFlowStatus;
@@ -41,13 +42,18 @@ import org.mockito.junit.jupiter.MockitoExtension;
import java.io.IOException;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
import static org.junit.jupiter.api.Assertions.assertEquals;
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;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -144,6 +150,26 @@ class StandardProcessGroupTest {
);
}
+ @Test
+ void testStartAndStopProcessingDoNothingWhenParentIsStateless() throws Exception {
+ // Callers that resolve a component's owning stateless group rely on this no-op: only the top-most stateless
+ // group may be started or stopped directly. It is also what terminates the recursive stopComponents() walk,
+ // so making it throw or delegate upward would break stateless shutdown.
+ when(parentProcessGroup.getExecutionEngine()).thenReturn(ExecutionEngine.STATELESS);
+ when(parentProcessGroup.resolveExecutionEngine()).thenReturn(ExecutionEngine.STATELESS);
+ processGroup.setParent(parentProcessGroup);
+ processGroup.setExecutionEngine(ExecutionEngine.STATELESS);
+
+ processGroup.startProcessing();
+
+ final CompletableFuture<Void> stopFuture = processGroup.stopProcessing();
+
+ assertTrue(stopFuture.isDone());
+ assertNull(stopFuture.get());
+ verify(processScheduler, never()).startStatelessGroup(any());
+ verify(processScheduler, never()).stopStatelessGroup(any());
+ }
+
@Test
void testGetLoggingAttributesWithoutParentProcessGroup() {
processGroup.setName(NAME);
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 6e375542796..362e1b814be 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<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);
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 36be3f53163..82df40c22ae 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.bundle.BundleCoordinate;
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.Set;
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.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,212 @@ class StandardControllerServiceProviderTest {
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 ProcessGroup inheritedGrandchild = createGroup(ExecutionEngine.INHERITED, inheritedChild);
+ final ProcessorNode childProcessor = createProcessor(inheritedGrandchild, 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(inheritedGrandchild, never()).stopProcessing();
+ verify(inheritedGrandchild, 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();
+ }
+
+ @Test
+ void testUnscheduleReferencingComponentsStopsTopMostStatelessGroupWhenNested() {
+ final ProcessGroup root = createGroup(ExecutionEngine.STANDARD, null);
+ final ProcessGroup inheritedChild = createGroup(ExecutionEngine.INHERITED, root);
+ final ProcessGroup outer = createGroup(ExecutionEngine.STATELESS, inheritedChild);
+ final ProcessGroup middle = createGroup(ExecutionEngine.STATELESS, outer);
+ final ProcessGroup inner = createGroup(ExecutionEngine.STATELESS, middle);
+
+ final ProcessorNode innerProcessor = createProcessor(inner, ScheduledState.RUNNING, ScheduledState.RUNNING);
+
+ when(outer.stopProcessing()).thenReturn(CompletableFuture.completedFuture(null));
+
+ final ControllerServiceNode serviceNode = createServiceWithProcessorReferences(List.of(innerProcessor));
+
+ final Map<ComponentNode, Future<Void>> result = serviceProvider.unscheduleReferencingComponents(serviceNode);
+
+ // stopProcessing() on a nested stateless group is a no-op that still returns a completed future, so stopping the
+ // wrong group would leave it running while the caller believes the stop succeeded.
+ verify(outer, times(1)).stopProcessing();
+ verify(middle, never()).stopProcessing();
+ verify(inner, never()).stopProcessing();
+ verify(inner, never()).stopProcessor(any());
+ assertTrue(result.containsKey(innerProcessor));
+ }
+
+ @Test
+ void testScheduleReferencingComponentsStartsTopMostStatelessGroupWhenNested() {
+ final ProcessGroup root = createGroup(ExecutionEngine.STANDARD, null);
+ final ProcessGroup inheritedChild = createGroup(ExecutionEngine.INHERITED, root);
+ final ProcessGroup outer = createGroup(ExecutionEngine.STATELESS, inheritedChild);
+ final ProcessGroup middle = createGroup(ExecutionEngine.STATELESS, outer);
+ final ProcessGroup inner = createGroup(ExecutionEngine.STATELESS, middle);
+
+ final ProcessorNode innerProcessor = createProcessor(inner, ScheduledState.STOPPED, ScheduledState.STOPPED);
+
+ final ControllerServiceNode serviceNode = createServiceWithProcessorReferences(List.of(innerProcessor));
+ final ComponentScheduler componentScheduler = mock(ComponentScheduler.class);
+
+ final Set<ComponentNode> result = serviceProvider.scheduleReferencingComponents(serviceNode, null, componentScheduler);
+
+ verify(componentScheduler, times(1)).startStatelessGroup(outer);
+ verify(componentScheduler, never()).startStatelessGroup(middle);
+ verify(componentScheduler, never()).startStatelessGroup(inner);
+ verify(componentScheduler, never()).startComponent(innerProcessor);
+ assertTrue(result.contains(innerProcessor));
+ }
+
+ private ProcessGroup createGroup(final ExecutionEngine executionEngine, final ProcessGroup parent) {
+ // Resolve before stubbing: reading the parent mock inside an in-progress when(...) trips Mockito's
+ // unfinished-stubbing detection.
+ final ExecutionEngine resolvedExecutionEngine = resolveExecutionEngine(executionEngine, parent);
+
+ final ProcessGroup group = mock(ProcessGroup.class);
+ lenient().when(group.getParent()).thenReturn(parent);
+ lenient().when(group.resolveExecutionEngine()).thenReturn(resolvedExecutionEngine);
+ return group;
+ }
+
+ /**
+ * Mirrors {@code StandardProcessGroup.resolveExecutionEngine()}. Callers must create a parent before its children
+ * so that the parent's stubbed resolution is already in place.
+ */
+ private ExecutionEngine resolveExecutionEngine(final ExecutionEngine executionEngine, final ProcessGroup parent) {
+ if (executionEngine != ExecutionEngine.INHERITED) {
+ return executionEngine;
+ }
+
+ return parent == null ? ExecutionEngine.STANDARD : parent.resolveExecutionEngine();
+ }
+
+ 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,
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 9eddbf597e0..5021f99dd2a 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
@@ -47,6 +47,7 @@ import java.util.Set;
* - A child process group with input and output ports
* - A processor within the child group
* - A stateless group with a processor
+ * - A pair of nested stateless groups, where only the inner one holds a processor
*
* This allows testing that start/stop operations properly handle all component types recursively.
*/
@@ -95,7 +96,7 @@ public class ComponentLifecycleConnector extends AbstractConnector {
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 +110,7 @@ public class ComponentLifecycleConnector extends AbstractConnector {
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,39 +132,85 @@ public class ComponentLifecycleConnector extends AbstractConnector {
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();
+ final VersionedProcessGroup nestedOuterGroup = createNestedStatelessGroups(childGroup.getIdentifier(), rootCountServiceId);
+ childGroup.getProcessGroups().add(nestedOuterGroup);
+
+ final VersionedPort nestedOuterInputPort = nestedOuterGroup.getInputPorts().iterator().next();
+
VersionedFlowUtils.addConnection(childGroup, VersionedFlowUtils.createConnectableComponent(inputPort),
VersionedFlowUtils.createConnectableComponent(childProcessor), Set.of(""));
VersionedFlowUtils.addConnection(childGroup, VersionedFlowUtils.createConnectableComponent(inputPort),
VersionedFlowUtils.createConnectableComponent(statelessInputPort), Set.of(""));
+ VersionedFlowUtils.addConnection(childGroup, VersionedFlowUtils.createConnectableComponent(inputPort),
+ VersionedFlowUtils.createConnectableComponent(nestedOuterInputPort), Set.of(""));
VersionedFlowUtils.addConnection(childGroup, VersionedFlowUtils.createConnectableComponent(childProcessor),
VersionedFlowUtils.createConnectableComponent(outputPort), Set.of("success"));
return childGroup;
}
- private VersionedProcessGroup createStatelessGroup(final String parentGroupId) {
- final VersionedProcessGroup statelessGroup = VersionedFlowUtils.createProcessGroup("stateless-group-id", "Stateless Group");
- statelessGroup.setPosition(new Position(400, 100));
- statelessGroup.setRemoteProcessGroups(new HashSet<>());
- statelessGroup.setScheduledState(ScheduledState.ENABLED);
- statelessGroup.setExecutionEngine(ExecutionEngine.STATELESS);
- statelessGroup.setStatelessFlowTimeout("1 min");
- statelessGroup.setGroupIdentifier(parentGroupId);
+ private VersionedProcessGroup createStatelessGroup(final String parentGroupId, final String rootCountServiceId) {
+ final VersionedProcessGroup statelessGroup = createStatelessGroupShell("stateless-group-id", "Stateless", new Position(400, 100), parentGroupId);
+ addCountingFlow(statelessGroup, "Stateless", rootCountServiceId);
+ return statelessGroup;
+ }
- final VersionedPort statelessInput = VersionedFlowUtils.addInputPort(statelessGroup, "Stateless Input", new Position(0, 0));
+ private VersionedProcessGroup createNestedStatelessGroups(final String parentGroupId, final String rootCountServiceId) {
+ final VersionedProcessGroup outerGroup = createStatelessGroupShell("nested-stateless-outer-group-id", "Nested Outer",
+ new Position(400, 300), parentGroupId);
- final VersionedProcessor statelessProcessor = VersionedFlowUtils.addProcessor(statelessGroup,
- "org.apache.nifi.processors.tests.system.TerminateFlowFile", SYSTEM_TEST_EXTENSIONS_BUNDLE, "Stateless Terminate", new Position(100, 100));
+ // Only the inner group holds a processor referencing the root service, so a resolver that stops at the inner
+ // group silently no-ops instead of transitioning the subtree.
+ final VersionedProcessGroup innerGroup = createStatelessGroupShell("nested-stateless-inner-group-id", "Nested Inner",
+ new Position(200, 0), outerGroup.getIdentifier());
+ addCountingFlow(innerGroup, "Nested Inner", rootCountServiceId);
+ outerGroup.getProcessGroups().add(innerGroup);
- VersionedFlowUtils.addConnection(statelessGroup, VersionedFlowUtils.createConnectableComponent(statelessInput),
- VersionedFlowUtils.createConnectableComponent(statelessProcessor), Set.of(""));
+ VersionedFlowUtils.addConnection(outerGroup, VersionedFlowUtils.createConnectableComponent(getInputPort(outerGroup)),
+ VersionedFlowUtils.createConnectableComponent(getInputPort(innerGroup)), Set.of(""));
- return statelessGroup;
+ return outerGroup;
+ }
+
+ private VersionedProcessGroup createStatelessGroupShell(final String identifier, final String namePrefix, final Position position,
+ final String parentGroupId) {
+ final VersionedProcessGroup group = VersionedFlowUtils.createProcessGroup(identifier, namePrefix + " Group");
+ group.setPosition(position);
+ group.setRemoteProcessGroups(new HashSet<>());
+ group.setScheduledState(ScheduledState.ENABLED);
+ group.setExecutionEngine(ExecutionEngine.STATELESS);
+ group.setStatelessFlowTimeout("1 min");
+ group.setGroupIdentifier(parentGroupId);
+
+ VersionedFlowUtils.addInputPort(group, namePrefix + " Input", new Position(0, 0));
+ return group;
+ }
+
+ /**
+ * Wires {@code input port -> CountFlowFiles -> TerminateFlowFile} inside the given group, with the CountFlowFiles
+ * processor referencing a Controller Service that lives outside the group.
+ */
+ private void addCountingFlow(final VersionedProcessGroup group, final String namePrefix, final String rootCountServiceId) {
+ final VersionedProcessor countProcessor = VersionedFlowUtils.addProcessor(group,
+ "org.apache.nifi.processors.tests.system.CountFlowFiles", SYSTEM_TEST_EXTENSIONS_BUNDLE, namePrefix + " Count", new Position(100, 50));
+ countProcessor.getProperties().put("Count Service", rootCountServiceId);
+
+ final VersionedProcessor terminateProcessor = VersionedFlowUtils.addProcessor(group,
+ "org.apache.nifi.processors.tests.system.TerminateFlowFile", SYSTEM_TEST_EXTENSIONS_BUNDLE, namePrefix + " Terminate", new Position(100, 100));
+
+ VersionedFlowUtils.addConnection(group, VersionedFlowUtils.createConnectableComponent(getInputPort(group)),
+ VersionedFlowUtils.createConnectableComponent(countProcessor), Set.of(""));
+ VersionedFlowUtils.addConnection(group, VersionedFlowUtils.createConnectableComponent(countProcessor),
+ VersionedFlowUtils.createConnectableComponent(terminateProcessor), Set.of("success"));
+ }
+
+ private VersionedPort getInputPort(final VersionedProcessGroup group) {
+ return group.getInputPorts().iterator().next();
}
@Override
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 1bbe86c3f21..0bc80db4391 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.AssetsEntity;
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.ProcessGroupEntity;
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,74 @@ import static org.junit.jupiter.api.Assertions.fail;
*/
public class ConnectorTroubleshootingIT extends NiFiSystemIT {
+ /**
+ * Regression test for stateless-group handling in the controller-service reference lifecycle. The managed flow has
+ * two stateless subtrees that both reference a controller service defined at the connector root: a flat one, and one
+ * where the referencing processor sits in a nested group that also declares the Stateless Engine. Stopping and
+ * starting the service's referencing components must transition each stateless subtree as a single unit, resolving
+ * up to the top-most stateless group rather than leaving 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);
+
+ final List<ProcessorEntity> statelessProcessors = findStatelessProcessors(connectorId);
+ assertEquals(4, statelessProcessors.size(), "Stateless groups should contain exactly four processors");
+ final List<String> 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<String> 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 a stateless group");
+ assertFalse(referencingProcessorIds.containsAll(statelessProcessorIds),
+ "At least one stateless processor should not directly reference the service, to prove group-as-unit behavior");
+
+ final ProcessorEntity nestedCountProcessor = findProcessorByName(connectorId, "Nested Inner Count");
+ assertNotNull(nestedCountProcessor, "Managed flow should contain the nested stateless Count processor");
+ assertTrue(referencingProcessorIds.contains(nestedCountProcessor.getId()),
+ "The nested-group Count processor should reference the root service, so top-most-group resolution is covered");
+
+ // Start the referencing components; every stateless subtree must start whole.
+ updateReferenceState(rootServiceId, references, ScheduledState.RUNNING.name());
+ for (final String processorId : statelessProcessorIds) {
+ waitForProcessorState(processorId, ScheduledState.RUNNING);
+ }
+
+ 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 +630,44 @@ public class ConnectorTroubleshootingIT extends NiFiSystemIT {
assertEquals(expected.name(), entity.getComponent().getState());
}
+ private List<ProcessorEntity> findStatelessProcessors(final String connectorId) throws NiFiClientException, IOException {
+ final List<ProcessorEntity> result = new ArrayList<>();
+ final Map<String, Boolean> 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<String, RevisionDTO> 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