[bobpaulin]

Start and stop each nested processor once when draining a Connector

startNonSourceComponents and stopSourceComponents collected processors recursively at every level while also recursing into child groups, so a processor N groups deep was started or stopped N+1 times. A repeated start
of a running processor never completes, which left drainFlowFiles blocked
and the Connector stuck in DRAINING. Each level now visits only its own processors, classified against the root group's connections.

Summary

NIFI-16459

Tracking

Please complete the following tracking steps prior to pull request creation.

Issue Tracking

• Apache NiFi Jira issue created

Pull Request Tracking

• Pull Request title starts with Apache NiFi Jira issue number, such as NIFI-00000
• Pull Request commit message starts with Apache NiFi Jira issue number, as such NIFI-00000

Pull Request Formatting

• Pull Request based on current revision of the main branch
• Pull Request refers to a feature branch with one commit containing changes

[markap14]

[Claude Opus 5.5] I believe this change is correct, but I think there's a simpler way to fix it. When called on the root group, getSourceProcessors and getNonSourceProcessors already return the right processors from the entire tree, classified against every connection in the flow. The port, Remote Process Group, and stateless group start methods also accept ComponentHierarchyScope.INCLUDE_CHILD_GROUPS. So we can drop the recursion completely. That removes the private overloads, the ID set passed down the tree, and the negated contains checks, and each processor is visited exactly once:

protected CompletableFuture<Void> stopSourceComponents(final FlowContext context) {
    final List<CompletableFuture<Void>> stopFutures = new ArrayList<>();
    for (final ProcessorFacade sourceProcessor : getSourceProcessors(context.getRootGroup())) {
        stopFutures.add(sourceProcessor.getLifecycle().stop());
    }

    return CompletableFuture.allOf(stopFutures.toArray(new CompletableFuture[0]));
}

protected CompletableFuture<Void> startNonSourceComponents(final FlowContext flowContext) {
    final ProcessGroupFacade rootGroup = flowContext.getRootGroup();
    final List<CompletableFuture<Void>> startFutures = new ArrayList<>();
    for (final ProcessorFacade nonSourceProcessor : getNonSourceProcessors(rootGroup)) {
        startFutures.add(nonSourceProcessor.getLifecycle().start());
    }

    final ProcessGroupLifecycle lifecycle = rootGroup.getLifecycle();
    startFutures.add(lifecycle.startPorts(ComponentHierarchyScope.INCLUDE_CHILD_GROUPS));
    startFutures.add(lifecycle.startRemoteProcessGroups(ComponentHierarchyScope.INCLUDE_CHILD_GROUPS));
    startFutures.add(lifecycle.startStatelessGroups(ComponentHierarchyScope.INCLUDE_CHILD_GROUPS));

    return CompletableFuture.allOf(startFutures.toArray(new CompletableFuture[0]));
}

In the framework's StandaloneProcessGroupLifecycle, the single root-level calls behave the same as one call per group, with one exception: groups nested inside a stateless group. The recursive startStatelessGroups call doesn't go into those. The per-group walk does go into them, and may call start() on them again. So the simpler version is likely more correct in that case as well. The new tests in this PR should still pass against this version.

[markap14]

Thanks @bobpaulin this looks correct. Looks like we were gathering recursively - recursively. Whoops! +1 will merge.