NIFI-16232 - Allow versioned flow updates to drain removed connections

[pvillard31]

Summary

NIFI-16232 - Allow versioned flow updates to drain removed connections

Versioned flow updates currently stop both endpoints of an affected connection before verifying that a connection removed by the target version has an empty queue. Stopping the destination prevents queued FlowFiles from draining, so an update can fail even when the destination could otherwise process the remaining data.

This change adds an Update-only pre-drain phase that:

• Identifies removed connections with non-empty queues.
• Conservatively rejects unsupported topologies before changing runtime state.
• Stops producer barriers while leaving destinations running.
• Waits up to 30 seconds for all eligible removed connections to drain.
• Requires exact queue-status coverage across connected cluster nodes.
• Restores components stopped during pre-drain following cancellation, timeout, or failure.
• Preserves the existing connection-removal verification and synchronization safety checks.

The approach follows the same lifecycle principle as Connector draining: stop accepting or producing new data, allow in-flight data to complete, and finish with components stopped before applying an update. It does not reuse the Connector drain API because Connector draining is component-owned and exposes a dedicated DRAINING state, while ordinary connections have no drain operation. Instead, versioned flow updates coordinate existing processor and port lifecycle operations and observe connection queues externally.

The behavior is limited to registry Update requests. Revert, rebase, process-group replacement, startup synchronization, and Connector update behavior are unchanged. Updates without non-empty removed connections complete the new progress step immediately.

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 contains commits signed with a registered key indicating Verified status

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

Verification

Please indicate the verification steps performed prior to pull request creation.

Build

• Build completed using ./mvnw clean install -P contrib-check
  • JDK 21
  • JDK 25

Licensing

• New dependencies are compatible with the Apache License 2.0 according to the License Policy
• New dependencies are documented in applicable LICENSE and NOTICE files

Documentation

• Documentation formatting appears as expected in rendered files

[markap14]

[claude-opus-4.8] Reviewed for correctness, thread safety, documentation, logging, and test coverage.

The design is sound and the classification logic is careful. I verified the three behaviors the feature depends on: destinations do stay running during the drain (only the producer-barrier set is ever passed to scheduleComponents(..., STOPPED)), producers stay stopped afterwards so the later empty-queue check remains meaningful, and unsupported topologies are rejected before anything is stopped. The cluster wait correctly refuses to treat a connection as drained when a node fails to report queue status.

My concerns are concentrated in the failure and cancellation paths of RemovedConnectionDrainCoordinator, where components that the framework stopped on its own initiative can be left stopped with no restore and no log entry. Details inline; the first two are the ones I would want addressed before merge.

Test coverage of what is implemented looks good. The one thing I would ask about is the clustered test that is disabled rather than fixed, since the cluster path holds the most complex new code.

[markap14]

[claude-opus-4.8] Producers can be left stopped with no restore when the drain wait throws an unchecked exception.

This catch handles only LifecycleManagementException, and the finally below only resets the cancel callback. But in a cluster the wait can throw unchecked exceptions that come straight from request replication.

ClusterReplicationComponentLifecycle.waitForConnectionQueuesEmpty calls createFlowFileListingRequest, which replicates with performVerification = true:

// ClusterReplicationComponentLifecycle line 789
private AsyncClusterResponse replicateFlowFileListingRequest(final Set<NodeIdentifier> expectedNodes, final NiFiUser user, final String method, final URI requestUri) {
    return getRequestReplicator().replicate(expectedNodes, user, method, requestUri, Collections.emptyMap(), Collections.emptyMap(), true, true);
}

ThreadPoolRequestReplicator throws two unchecked exceptions out of that call:

• IllegalClusterStateException (line 365) when a node in expectedNodes is no longer CONNECTED.
• ConnectingNodeMutableRequestException (line 708, via verifyClusterState) when any node is CONNECTING. This one applies because creating a listing request is a POST, so it is classified as a mutable request.

Both extend ClusterException and are unchecked. createFlowFileListingRequest only catches InterruptedException, so they propagate out of the wait and past this catch block.

The target node set is a snapshot taken once per poll iteration at line 646, so an ordinary membership change during the 30-second drain is enough to trigger this. FlowUpdateResource line 307 then catches Exception and fails the request, but nothing restores the producers. The user is left with a failed update and a set of processors the framework stopped on its own and never restarted, with nothing in nifi-app.log saying so.

Suggest catching Exception (or at minimum ClusterException) here, or moving the restore into the finally so it runs regardless of how the try block exits.

[markap14]

[claude-opus-4.8] Cancellation is ignored when the queues drain successfully, and the drain-stopped producers are not restored.

The isCancelled() check at line 131 only runs when the wait returns false. ClusterReplicationComponentLifecycle.waitForConnectionQueuesEmpty returns true at line 672 as soon as it observes all queues empty, without consulting the pause. So if the user cancels the request in the same moment the queues drain, this returns success.

What follows in FlowUpdateResource:

1. Line 368 merges the drain-stopped producers into runningComponents.
2. Line 375 stops the full affected set, including the destinations the coordinator had deliberately left running.
3. Line 377 returns because the request is cancelled, which happens before the try/finally at lines 429/489 that would restart anything.

Both the producers and the destinations end up stopped. NiFi already leaves components stopped when a user cancels mid-stop, so that part is not new, but these particular producers were stopped by the framework rather than by the user's own step, and the documented contract of this class is to restore them on cancellation.

Checking cancellationHandle.isCancelled() before returning success, and routing to restoreAfterCancellation when it is set, would make the contract hold. Worth a coordinator unit test for cancel-during-a-successful-wait; the existing cancellation test only covers the case where the wait returns false, which is exactly why this path slipped through.

[markap14]

[claude-opus-4.8] No logging anywhere in the drain path.

Neither this class nor RemovedConnectionDrainClassifier declares a logger.

From an operator's point of view, the framework stops a set of processors that the user did not ask to stop, waits up to 30 seconds, and then either proceeds or restores them, and nifi-app.log contains nothing about any of it. When an update takes an unexplained 30 extra seconds or fails with a drain timeout, there is no record of which connections were being drained, which producers were stopped, how much data remained, or whether restoration succeeded.

Suggested minimum:

• INFO when the drain begins, naming the connections to drain and the producers being stopped.
• INFO or WARN on timeout with the remaining queue sizes, since that is the case an operator will actually investigate.
• WARN when restoration fails, from this class. Today the only such log is FlowUpdateResource line 363, which covers the cancellation path only; the restoration failure in the catch block at line 136 is attached to the thrown exception and never logged here.
• DEBUG per poll iteration in waitForConnectionQueuesEmpty showing which nodes reported and which queues are still non-empty. Without this, diagnosing an incomplete-coverage timeout in a cluster means attaching a debugger.

[markap14]

[claude-opus-4.8] FUNNEL_SOURCE_WITHOUT_SUPPORTED_PRODUCER is reported whenever no producer barrier can be resolved, not only for funnel sources. A connection whose source is a remote process group port, or a processor whose upstream traversal terminates unexpectedly, is reported to the user as a funnel problem.

This matters because the reason name reaches the user: buildClassificationFailureMessage puts it into the request's failure reason, which surfaces in the UI.

Either rename the constant to something accurate such as NO_SUPPORTED_PRODUCER_FOUND, or split the funnel case out from the general one so the funnel-specific reason is only used when the source really is a funnel.

[markap14]

[claude-opus-4.8] A connection whose group cannot be resolved is treated as retained.

When processGroupId is null the while loop at line 320 never executes and this returns true, meaning "this group survives the update", which makes the connection a drain candidate.

This is reachable rather than defensive: StandardNiFiServiceFacade.getComponentGroupRuntimeId can return null.

Every other ambiguous case in this class resolves toward unsupported, which is the safe direction because it fails the request before anything is stopped. Note that the not-found case a few lines up at line 330 already returns false. Here an unresolvable group instead leads to stopping producers for a connection whose containing group may itself be going away, in which case the queue never drains and the request burns the full 30 seconds before timing out.

Returning false for a null group id, to match the not-found case, would keep this consistent with the rest of the class.

[markap14]

[claude-opus-4.8] "update-requests" is a bare string literal here, and again at line 644 in getUpdateFlowSteps, while VersionsResource already defines constants for request types and passes this same value at line 1426.

Since both the drain step itself and the progress-step list are gated on this exact string, a typo in a future caller silently disables draining rather than failing, and the two gates can drift apart from each other. Worth promoting to a shared constant alongside the existing request-type constants.

[markap14]

[claude-opus-4.8] The numbered workflow description in the comment block starting at line 175 still describes the old sequence and does not mention the new pre-drain step, even though the step was inserted into the middle of the flow it documents.

That comment is the first thing a maintainer reads before touching this method, and the step numbering it establishes (// Step 1: here, // Steps 5-6: further down) is now out of sync with what the code actually does. Worth updating it to include the drain step.

[markap14]

[claude-opus-4.8] This throws NullPointerException when the request has not already failed.

Objects.requireNonNull(failureReason, ...) makes "a failure reason already exists" a hard precondition, but nothing in the AsynchronousWebRequest.appendFailureDetail signature communicates that, and the interface method has no Javadoc stating it.

The single caller today, FlowUpdateResource line 364, happens to satisfy the precondition because cancellation always sets a failure reason first. But the method is on a public framework interface, so the next caller has no way to know the ordering requirement short of reading this implementation, and gets an NPE if they get it wrong. An NPE is a bug, not an API contract.

Suggest one of: append the detail to an empty base when no failure reason is set, or throw IllegalStateException with a message that explains the ordering requirement. Either way the interface method should say what it requires.

[markap14]

[claude-opus-4.8] If stopConnector throws, the test fails either way, so the assertDoesNotThrow wrapper adds a lambda and an import without adding any signal. The same applies at line 85. Just calling testRunner.stopConnector(Duration.ofSeconds(120)) directly reads better and behaves identically.

Separately, this change appears unrelated to the removed-connection drain work. It would be easier to review, and easier to revert independently, as its own commit.

[markap14]

[claude-opus-4.8] Could you say more about why this fails in a cluster? The disable reason describes the fixture mismatch (the inherited test waits for a single port-output FlowFile while clustered GenerateFlowFile queues per node), but that reads like a reason the test needs a cluster-aware fixture rather than a reason the scenario cannot be covered.

This matters because the cluster path holds the most complex new code in the PR: request replication, the node-coverage check, and the mutable-request verification I flagged in RemovedConnectionDrainCoordinator. Disabling this leaves that path with unit coverage only, at the same time as the PR removes the one clustered end-to-end exercise of it.

If the underlying issue is only the FlowFile-count assertion, adapting the fixture to expect per-node counts seems worth doing here. If it is something deeper about how the drain behaves in a cluster, that is worth understanding before merge, since it may be a finding rather than a test-infrastructure problem.

[markap14]

[claude-opus-4.8] Marking the review I just left as Request Changes. The detailed comments are in #11570 (review); this is only to set the review state correctly, not to add new findings.

The two items I consider blocking are both in RemovedConnectionDrainCoordinator, and both leave components that the framework stopped on its own initiative in a stopped state with no restore and nothing in the log:

1. Line 136 catches only LifecycleManagementException, but in a cluster the drain wait replicates a POST with performVerification = true and can throw IllegalClusterStateException or ConnectingNodeMutableRequestException, both unchecked. An ordinary cluster membership change during the 30-second drain is enough to trigger it, and the producers are never restarted.

2. Line 127 returns success without checking isCancelled(). If cancellation lands on the poll that observes the queues empty, FlowUpdateResource goes on to stop the destinations too and then returns before reaching the block that would restart anything, so producers and destinations are both left stopped even though this class is documented to restore on cancellation.

The remaining comments (classifier reason naming, null group id treated as retained, the appendFailureDetail NPE, missing logging, the request-type literal, the stale workflow comment, and the disabled clustered test) are worth addressing but I would not hold the PR on them individually.

The overall approach is good and the classification logic is careful. I verified that destinations stay running during the drain, that producers stay stopped so the later empty-queue check remains meaningful, that unsupported topologies are rejected before anything is stopped, and that the cluster wait does not treat a non-reporting node as drained.