Bob Paulin created NIP-49:
-----------------------------
Summary: 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: Task
Reporter: Bob Paulin
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*
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();
}
The context provides the pre-upgrade flow, the loaded Connector bundle, and the
connections the upgraded flow will remove.
h3. *Connector.prepareForFlowEvolution*
default void prepareForFlowEvolution(FlowEvolutionContext context) throws
FlowUpdateException {
}
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)
