[
https://issues.apache.org/jira/browse/NIP-49?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Mark Payne resolved NIP-49.
---------------------------
Resolution: Fixed
> Connector Upgrade Drain and Flow Definition Persistence
> -------------------------------------------------------
>
> Key: NIP-49
> URL: https://issues.apache.org/jira/browse/NIP-49
> Project: NiFi Improvement Proposal
> Issue Type: Improvement
> Reporter: Bob Paulin
> Priority: Major
> Attachments: image-2026-10-12-08-39-24-494.png
>
>
> This proposal complements NIP-47. NIP-47 relocates FlowFiles after a flow has
> been replaced. This proposal covers the step before that: persisting the flow
> a connector last applied, restoring it when different Connector code is
> loaded, and giving the connector a chance to empty connections the upgrade
> will remove before the new flow is installed.
> h2. Motivation
> A Connector version change can load different code and rebuild the managed
> flow from configuration. Whenever the loaded version differs from the version
> that built the authoritative flow, the framework must detect the change and
> perform flow evolution before starting the Connector.
> Changing a Connector version at runtime is not currently supported, but may
> be supported in the near future. This proposal does not define how or when a
> version is selected and does not introduce runtime version changes. It
> defines one persistence and evolution contract that applies whenever the
> loaded version changes.
> Without that contract, connections from the previous flow can disappear and
> any FlowFiles still queued in them become orphaned. A Connector that needs to
> finish processing that data, or discard it deliberately, has no hook that
> runs while the previous flow is still installed.
> The previous flow also is not written to {{flow.json}} except while the
> Connector is in troubleshooting. When different code is loaded, the framework
> rebuilds the flow by calling that code, so the queues have nothing to
> reattach to.
> h2. *Proposed nifi-api changes*
> All additions are backward compatible. Existing connectors compile and run
> unchanged.
> h3. *FlowEvolutionContext*
> {code:java}
> public interface FlowEvolutionContext {
> /**
> * The pre-upgrade Active flow. Its flow, parameters, and bundle are from
> the authoritative snapshot.
> * Its configuration has already been migrated and filtered for the
> loaded Connector version.
> */
> FlowContext getPreUpgradeActiveFlowContext();
> /** The bundle of the loaded Connector code, whether loaded at startup or
> changed at runtime. */
> Bundle getPostUpgradeBundle();
> }
> {code}
> The context provides the pre-upgrade flow and the loaded Connector bundle. If
> flow evolution fails, the framework replaces the supplied context with a
> wrapper that rejects every method declared by {{{}FlowEvolutionContext{}}},
> including {{getPreUpgradeActiveFlowContext()}} and
> {{{}getPostUpgradeBundle(){}}}. This covers timeout, cancellation, and any
> other evolution failure, so a connector that retained the context cannot
> continue using it.
> h3. *Connector.prepareForFlowEvolution*
> {code:java}
> default CompletableFuture<Void> prepareForFlowEvolution(FlowEvolutionContext
> context) throws FlowUpdateException {
> return CompletableFuture.completedFuture(null);
> } {code}
> The framework invokes this hook while the pre-upgrade flow is still installed
> and before the upgraded flow removes connections. The method returns promptly
> with a non-null future that owns all preparation work. The hook empties
> pre-upgrade connections the upgraded flow no longer contains before that
> future completes. Synchronous setup can throw {{{}FlowUpdateException{}}};
> asynchronous failures complete the future exceptionally. The hook is not
> called when the Connector leaves Troubleshooting; the managed flow must
> already be empty then.
> The framework cancels the future on evolution timeout, user drain
> cancellation or stop, coordinator interruption, and cluster monitoring
> failure. {{CompletableFuture.cancel(true)}} does not interrupt arbitrary
> executor tasks, so connector implementations must make delegated work
> cooperatively cancellable and must not complete the future while
> fire-and-forget work can still mutate the pre-upgrade flow. Connector-created
> asynchronous threads must preserve the Connector NAR classloader context.
> Only normal completion advances phase one. The framework then predicts the
> upgraded flow with {{getActiveFlow}} and the Working context, and fails
> evolution if a removed connection still holds FlowFiles.
> h3. *Connector.getActiveFlow*
> The method must remain idempotent and build the flow only from the supplied
> context's configuration. When called with the Working context, its result
> predicts what {{applyUpdate}} will install.
> h3. *VersionedConnector*
> * managedProcessGroup, managedParameterContext, and managedFlowBundle
> persist the authoritative flow, its parameters, and the Connector version
> that built it.
> * troubleshootingProcessGroup separately persists user edits made in
> troubleshooting.
> * bundle identifies the Connector code to load and remains independent from
> managedFlowBundle.
> ConnectorInitializationContext is unchanged. It does not gain a bundle getter.
> h2. *Framework behavior*
> h3. *Persistence*
> VersionedComponentFlowMapper.mapConnector writes:
> * {{{}managedParameterContext{}}}, {{{}managedFlowBundle{}}}, and
> {{activeFlowConfiguration}} from one immutable in-memory
> {{{}ConnectorFlowSnapshot{}}}, never by combining values from different
> lifecycle phases. The record is a complete Connector flow-persistence value
> that can be prepared, published as authoritative, or retained for rollback.
> * {{managedProcessGroup}} from that snapshot, except while the connector is
> in Troubleshooting, when it is written from the live managed group.
> * {{bundle}} from the loaded Connector coordinate and {{managedFlowBundle}}
> from the snapshot. While an upgrade is pending, these fields differ.
> The authoritative snapshot is set after every successful apply, after an
> upgraded snapshot is committed, and when a persisted authoritative flow is
> loaded. {{getAuthoritativeFlowSnapshot()}} names the currently committed
> instance; prepared and rollback snapshots use the same
> {{ConnectorFlowSnapshot}} value type. It is captured on entering
> troubleshooting only when none is held. While troubleshooting, the persisted
> managed group is the live edits and the snapshot bundle stays the version
> that built the flow. On exiting troubleshooting, the snapshot is refreshed
> from the installed Active flow, including when an upgrade was pending.
> h3. *Detection*
> During synchronization, the framework resolves the snapshot bundle from
> {{{}managedFlowBundle{}}}, or infers it once from the persisted {{bundle}}
> when a managed group exists, including a group persisted while
> Troubleshooting. The inferred value is frozen before the loaded version can
> change. No value is inferred when there is no managed group.
> Troubleshooting is restored from {{managedProcessGroup}} and defers evolution
> until the Connector exits troubleshooting. The same detection rule applies
> whenever the loaded version changes.
> h3. *Replacing the flow*
> Each node waits for the {{prepareForFlowEvolution}} future before applying
> the working configuration and preparing the incoming snapshot. A timeout or
> cancellation cancels the future, retains the authoritative pre-upgrade
> snapshot, and leaves the Connector stopped and pending. After cluster-wide
> commit authorization, every node atomically publishes the incoming snapshot
> and persists its own flow, and START waits for cluster-wide persistence
> completion before proceeding to running. Publishing the new snapshot makes
> its managed-flow bundle equal the loaded bundle, so pending becomes false
> without clearing a stored flag.
> If preparing or committing the upgraded flow fails, the outgoing snapshot
> remains authoritative or is restored, and the node stays stopped and pending.
> The existing flow-update verification remains as a backstop for a FlowFile
> that arrives after the check, or for an {{applyUpdate}} that installs
> something other than {{getActiveFlow}} on the working context.
> *Exiting troubleshooting while an upgrade is pending*
> {{prepareForFlowEvolution}} is not run. Every queue in the managed flow,
> including child groups, must be empty. The exit is rejected while any
> FlowFiles remain, and the Connector stays in Troubleshooting with the upgrade
> still pending. When the queues are empty, the same synchronous exit used
> otherwise installs {{{}getActiveFlow{}}}, captures the snapshot with the
> loaded bundle, and leaves the Connector {{STOPPED}} and no longer pending.
> This covers a connector that was in troubleshooting when the runtime was
> upgraded, and a connector whose drain was cancelled so the user could enter
> troubleshooting from {{{}STOPPED{}}}.
> The hook is skipped because the empty-queue check already makes the
> replacement safe, and the shipped hooks only drain. Draining on exit would
> start components the user added and stopped, against a flow whose structure
> the hook cannot rely on after those edits. The user can empty the queues
> before exiting, and the exit stays a synchronous verify-then-apply across the
> cluster.
> h2. *New configuration*
> nifi.components.connectors.flow.evolution.timeout bounds
> prepareForFlowEvolution. The default is 30 mins.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
