[
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. 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 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:
* 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.
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.
> 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 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:
> * 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)
