[
https://issues.apache.org/jira/browse/NIP-49?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18126327#comment-18126327
]
Mark Payne commented on NIP-49:
-------------------------------
Hey [~bob] I think this approach makes sense. I'm happy to sponsor it as a PMC.
A couple of questions that are probably worth calling out specifically:
* When would the {{prepareForFlowEvolution}} method get called? I.e., what
does the lifecycle look like?
* How do we handle failures where one node succeeds but another node fails?
* 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.
* 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.
> 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
>
> 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)
