[
https://issues.apache.org/jira/browse/NIP-49?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Bob Paulin updated NIP-49:
--------------------------
Description:
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.
h3. *Connector.prepareForFlowEvolution*
{code:java}
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. The hook empties pre-upgrade
connections the upgraded flow no longer contains. After it returns, the
framework 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:
* 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.
was:
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.
> 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.
> h3. *Connector.prepareForFlowEvolution*
> {code:java}
> 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. The hook empties
> pre-upgrade connections the upgraded flow no longer contains. After it
> returns, the framework 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:
> * 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)
