[
https://issues.apache.org/jira/browse/NIP-49?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18126390#comment-18126390
]
Bob Paulin commented on NIP-49:
-------------------------------
Hi [~markap14] thank you for the feedback. Responses inline.
* When would the {{prepareForFlowEvolution}} method get called? I.e., what
does the lifecycle look like?
The evolution would be at the beginning of the Connector START which could
either be auto resumed or user driven.
!image-2026-10-12-08-39-24-494.png!
* How do we handle failures where one node succeeds but another node fails?
When the evolve is triggered each node needs to run through the process so they
all need to succeed for the evolution to be considered successful. In the
event of one node failing each node restores the old flow, sets the desired
state to {{{}STOPPED{}}}, and leaves the evolution pending. None of them
install the upgraded flow or enter {{{}RUNNING{}}}. So the evolution must be
retried until it passes on all nodes in the cluster.
* I wonder if we actually need the {{getRemovedConnections()}} method. We have
the {{FlowContext getPreUpgradeActiveFlowContext()}} to determine what
connections exist already in the Connector. And the Connector itself is the
authority on what the flow looks like. So I'm not fully understanding why we
would have the framework determine this and then provide it as a set of
"predicted" connections to be removed. I might be missing something.
I think the baseline requirement should be we should not allow upgrades to
progress if a connection removal orphans flow files. To make that
determination prior to ditching the persisted flow requires the new flow to be
calculated while we still have both flows. So the design I'm promoting can
have this information prior to prepareForFlowEvolution. I had imagined while
the flow developer should know which connections would be removed since it
might not always be obvious. For example if a condition change in the code
that builds the flow were modified the connection might be programmatically
removed. So providing the developer with the expected removed connections
might help this process and allow the developer to avoid drains when the queues
of those connections are empty. All this said it is not critical to the design
and is a convenience that can be removed. We could consider it later if we
find evolutions are successful but developers are miscalculating connection
removals.
* Might be worth including as part of this effort some methods on the
{{ConnectorTestRunner}} / {{{}StandardConnectorTestRunner{}}}. Of course this
could come as a follow-on but worth thinking through how this might affect this
area, since we'll need to be able to test this well.
Yes agree. Including the ability to allow the Test Runner to artificially
trigger a pending evolution. I'd expect a lot of functionality to enqueue
FlowFiles on existing connections could be used to put the connector in a state
that requires drain and assert that it is in good state following the flow
evolution hook. So I expect this to be just some light additions and I can
include in the NIP if you feel its valuable.
> 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.
> Every version-change mechanism therefore uses the same detection, hook,
> validation, cluster barriers, atomic snapshot commit, persistence barrier,
> and normal start path.
>
> h2. *Proposed nifi-api changes*
> All additions are backward compatible. Existing connectors compile and run
> unchanged.
> h3. *FlowEvolutionContext*
> {code}
> 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();
> /** Connections the predicted upgraded flow removes, including
> connections in removed child groups. */
> Set<ConnectionFacade> getRemovedConnections();
> }
> {code}
> The context provides the pre-upgrade flow, the loaded Connector bundle, and
> the connections the upgraded flow will remove.
> h3. *Connector.prepareForFlowEvolution*
> {code}
> default void prepareForFlowEvolution(FlowEvolutionContext context) throws
> FlowUpdateException {
> }
> {code}
> The framework invokes this hook while the pre-upgrade flow is still installed
> and before the upgraded flow removes connections. Implementations use it to
> empty getRemovedConnections(); evolution fails if any of those connections
> still contains FlowFiles when the method returns.
> 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:
> * managedProcessGroup, 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.
> * bundle from the loaded Connector coordinate and managedFlowBundle from the
> snapshot. While an upgrade is pending, these fields differ.
> * troubleshootingProcessGroup from the live managed group, and only while
> the connector is in troubleshooting.
> 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. It is left empty after a legacy troubleshooting read, and it is not
> filled from edits made in troubleshooting. On exiting troubleshooting, when
> evolution is not pending, the authoritative snapshot is refreshed from the
> restored Active flow.
> h3. *Detection*
> During synchronization, the framework resolves the snapshot bundle from
> managedFlowBundle, or infers it once from the persisted bundle when an
> authoritative managed group exists. The inferred value is frozen before the
> loaded version can change. No value is inferred when there is no managed
> group or when that group is a legacy troubleshooting snapshot.
> Troubleshooting state is restored on top of the authoritative flow and defers
> evolution until the Connector exits troubleshooting. The same detection rule
> applies whenever the loaded version changes.
> h3. *Replacing the flow*
> Each node applies the working configuration and prepares the incoming
> snapshot during evolution. After cluster-wide commit authorization, every
> node atomically publishes it 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.
> 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)
