markap14 commented on code in PR #11240:
URL: https://github.com/apache/nifi/pull/11240#discussion_r3615110614
##########
nifi-docs/src/main/asciidoc/administration-guide.adoc:
##########
@@ -3852,6 +3852,93 @@ The Secrets Manager delegates to Parameter Providers to
retrieve secret values.
|`nifi.secrets.manager.cache.duration`|The duration for which resolved secret
values are cached before being refreshed from the underlying Parameter
Providers. Accepts any NiFi time duration value such as `5 mins`, `30 secs`,
etc. A value of `0 sec` disables caching entirely. Defaults to `5 mins`.
|====
+== Migrating a Connector from a Versioned Process Group
+
+NiFi supports a one-time path that migrates a newly created Connector from a
source Versioned Process Group.
+The source Versioned Process Group is used as a reference: the Connector reads
it and updates its own managed flow to mirror the source's configuration,
parameters, and component state.
+The source flow itself is not installed onto the Connector.
+Only Connectors that explicitly opt in to migration support can use this
capability.
+
+=== Prerequisites for a local migration
+
+When the source is a local Process Group on the same NiFi instance, the
Process Group must satisfy all of the following requirements before NiFi will
allow the migration:
+
+- The source Process Group must be under version control.
+- The source Process Group must be `UP_TO_DATE` with the latest published
version in Flow Registry.
+- All processors in the source Process Group must be stopped.
+- All controller services in the source Process Group must be disabled.
+- All queues in the source Process Group must be empty.
+- The source Process Group must not reference controller services outside of
the Process Group.
+- The target Connector must be newly created, stopped, and not previously
migrated.
+- The target Connector's active flow must be empty when migration begins.
Connectors that ship a non-empty initial flow are not compatible with the
migration path.
+
+If the source Process Group is not under version control, export it and use
the uploaded payload path instead.
+
+=== Migration paths
+
+There are two ways to run the migration:
+
+- Local Process Group migration:
+ Select an eligible Process Group that already exists on the current NiFi
canvas.
+ NiFi validates the prerequisites listed above before starting the migration.
+- Uploaded payload migration:
+ Export a Process Group definition and upload the resulting flow snapshot
payload to the Connector migration endpoint.
Review Comment:
[Sonnet 5] Good call — updated the uploaded payload migration step to
mention exporting component state.
##########
nifi-docs/src/main/asciidoc/administration-guide.adoc:
##########
@@ -3852,6 +3852,93 @@ The Secrets Manager delegates to Parameter Providers to
retrieve secret values.
|`nifi.secrets.manager.cache.duration`|The duration for which resolved secret
values are cached before being refreshed from the underlying Parameter
Providers. Accepts any NiFi time duration value such as `5 mins`, `30 secs`,
etc. A value of `0 sec` disables caching entirely. Defaults to `5 mins`.
|====
+== Migrating a Connector from a Versioned Process Group
+
+NiFi supports a one-time path that migrates a newly created Connector from a
source Versioned Process Group.
+The source Versioned Process Group is used as a reference: the Connector reads
it and updates its own managed flow to mirror the source's configuration,
parameters, and component state.
+The source flow itself is not installed onto the Connector.
+Only Connectors that explicitly opt in to migration support can use this
capability.
+
+=== Prerequisites for a local migration
+
+When the source is a local Process Group on the same NiFi instance, the
Process Group must satisfy all of the following requirements before NiFi will
allow the migration:
+
+- The source Process Group must be under version control.
+- The source Process Group must be `UP_TO_DATE` with the latest published
version in Flow Registry.
+- All processors in the source Process Group must be stopped.
+- All controller services in the source Process Group must be disabled.
+- All queues in the source Process Group must be empty.
+- The source Process Group must not reference controller services outside of
the Process Group.
+- The target Connector must be newly created, stopped, and not previously
migrated.
+- The target Connector's active flow must be empty when migration begins.
Connectors that ship a non-empty initial flow are not compatible with the
migration path.
+
+If the source Process Group is not under version control, export it and use
the uploaded payload path instead.
+
+=== Migration paths
+
+There are two ways to run the migration:
+
+- Local Process Group migration:
+ Select an eligible Process Group that already exists on the current NiFi
canvas.
+ NiFi validates the prerequisites listed above before starting the migration.
+- Uploaded payload migration:
+ Export a Process Group definition and upload the resulting flow snapshot
payload to the Connector migration endpoint.
+ This path is intended for non-version-controlled sources, sources exported
from another NiFi instance, or any other source that cannot satisfy the local
version-control requirements.
+
+For uploaded payloads, NiFi does not require the source flow to be under
version control.
+The Connector's own migration support check determines whether the uploaded
flow can be imported.
+
+=== Running a migration
+
+. Create a new Connector instance.
+. If the source Process Group is local and registry-backed, choose it from the
Connector migration source list.
+. If the source Process Group is not eligible for the local path, export it
with component state included and upload the payload to the Connector migration
payload endpoint.
+. Start the Connector migration request.
+. Poll the request until it completes.
+. Open the Connector configuration and supply any missing sensitive values or
secrets.
Review Comment:
[Sonnet 5] Updated as suggested.
##########
nifi-docs/src/main/asciidoc/administration-guide.adoc:
##########
@@ -3852,6 +3852,93 @@ The Secrets Manager delegates to Parameter Providers to
retrieve secret values.
|`nifi.secrets.manager.cache.duration`|The duration for which resolved secret
values are cached before being refreshed from the underlying Parameter
Providers. Accepts any NiFi time duration value such as `5 mins`, `30 secs`,
etc. A value of `0 sec` disables caching entirely. Defaults to `5 mins`.
|====
+== Migrating a Connector from a Versioned Process Group
+
+NiFi supports a one-time path that migrates a newly created Connector from a
source Versioned Process Group.
+The source Versioned Process Group is used as a reference: the Connector reads
it and updates its own managed flow to mirror the source's configuration,
parameters, and component state.
+The source flow itself is not installed onto the Connector.
+Only Connectors that explicitly opt in to migration support can use this
capability.
Review Comment:
[Sonnet 5] Updated as suggested.
##########
nifi-docs/src/main/asciidoc/developer-guide.adoc:
##########
@@ -2697,6 +2697,126 @@ The following command is used to generate a standard
binary distribution of Apac
`mvn clean install -Pcontrib-check`
+== Supporting migration from a Versioned Process Group
+
+Connectors can opt in to a one-time migration flow that uses a source
Versioned Process Group as a reference for populating a freshly created
Connector.
Review Comment:
[Sonnet 5] Updated as suggested.
##########
nifi-docs/src/main/asciidoc/developer-guide.adoc:
##########
@@ -2697,6 +2697,126 @@ The following command is used to generate a standard
binary distribution of Apac
`mvn clean install -Pcontrib-check`
+== Supporting migration from a Versioned Process Group
+
+Connectors can opt in to a one-time migration flow that uses a source
Versioned Process Group as a reference for populating a freshly created
Connector.
+This is intended for Connectors that can translate an exported flow
definition, its parameter contexts, referenced assets, and any attached
component state into their own managed flow.
+The source Versioned Process Group is not installed onto the Connector;
instead, the Connector reads the source as input and updates its own flow to
mirror the source's configuration and state.
+
+Migration support is optional.
+Connectors opt in by implementing the optional capability interface
`org.apache.nifi.components.connector.migration.MigratableConnector`.
+A Connector that does not implement `MigratableConnector` is never offered as
a migration target and continues to behave as before.
+
+=== Migration contract
+
+`MigratableConnector` is a separate interface that the Connector class must
implement (typically alongside extending `AbstractConnector`). It exposes three
methods:
+
+- `isMigrationSupported(ConnectorMigrationContext)` is a cheap,
side-effect-free predicate.
+ It should inspect `context.getSourceFlow()` and return `true` only when the
Connector can be migrated from the source structure.
+ Implementations should limit themselves to structural checks such as
processor types, parameter names, and exported metadata.
+ Component state and asset content are not available at listing time and must
not be relied on; per-state precondition checks belong inside
`migrateConfiguration` or `migrateState`.
+ This method must not call
`ConnectorMigrationContext.copyAssetFromSource(...)` or any of the write APIs
described below; the framework wraps the eligibility-time context so any such
attempt is rejected with `IllegalStateException` and the Connector is filtered
out of the listing.
+- `migrateConfiguration(ConnectorMigrationContext)` records the configuration
the Connector wants on the other side of the migration.
+ The Connector reads `context.getSourceFlow()` and records configuration
changes via `context.setProperties(stepName, propertyValues)` (merge) or
`context.replaceProperties(stepName, propertyValues)` (full replacement of the
step).
+ These calls are applied to a working copy of the Connector's configuration
that the framework seeded with the Connector's current active configuration, so
the working copy always reflects the fully-merged result.
+ After `migrateConfiguration` returns, the framework drives
`Connector.applyUpdate(workingContext, activeContext)` from that merged
configuration so the managed Process Group is rebuilt by the same code path the
framework uses on restart.
+ The Connector must not call `updateFlow(...)` from this method; the
framework hands the Connector a read-only wrapper around the active
`FlowContext`, so any direct `updateFlow(...)` attempt is rejected with
`IllegalStateException`.
+ Component-state writes are also rejected in this phase:
`setComponentState(...)` throws because the managed components do not exist yet.
+- `migrateState(ConnectorMigrationContext)` runs after the framework has
rebuilt the managed Process Group from the merged configuration.
+ At this point every Processor and Controller Service in the source flow has
a corresponding managed component in the active `FlowContext`, so the Connector
can match each source `VersionedConfigurableExtension` to its managed
counterpart by versioned identifier.
+ The Connector records the desired `StateManager` state for each managed
component via `context.setComponentState(versionedComponentId,
versionedComponentState)`.
+ After `migrateState` returns, the framework writes each staged
`VersionedComponentState` into the corresponding managed component's live
`StateManager`.
+ Configuration writes (`setProperties`, `replaceProperties`) and direct
`updateFlow(...)` calls are rejected in this phase for the same reasons
described above.
+
+The two phases together form a single, all-or-nothing migration. The framework
only commits the merged configuration onto the Connector's persisted active
configuration after both `migrateState` and the framework's subsequent
state-application step have succeeded. Any failure in either phase rolls the
migration back: the managed Process Group is restored from `getInitialFlow()`,
the working configuration copy is discarded, any assets copied via
`copyAssetFromSource(...)` are deleted, and the Connector's persisted
configuration remains unchanged.
+
+=== Accessing the source flow
+
+`ConnectorMigrationContext` exposes the source `VersionedExternalFlow` using
`getSourceFlow()`.
+The source includes:
+
+- The exported `VersionedProcessGroup`
+- Parameter contexts
+- Referenced assets
+- Attached `VersionedComponentState` for `@Stateful` components
+- Export metadata such as flow name, registry identifiers, and version
+
+The same context also indicates whether the request came from a local Process
Group on the same NiFi instance or from an uploaded payload using
`isLocalMigration()`.
+
+The active `FlowContext` is available via `context.getActiveFlowContext()` and
is read-only during migration: it forwards reads (configuration, managed
Process Group) to the live active flow context but rejects `updateFlow(...)`
with `IllegalStateException`. The Connector should treat this object as a
window onto the current state of the Connector, not as a write target.
Review Comment:
[Sonnet 5] Added a mention of applyUpdate to this section as well.
##########
nifi-docs/src/main/asciidoc/developer-guide.adoc:
##########
@@ -2697,6 +2697,126 @@ The following command is used to generate a standard
binary distribution of Apac
`mvn clean install -Pcontrib-check`
+== Supporting migration from a Versioned Process Group
+
+Connectors can opt in to a one-time migration flow that uses a source
Versioned Process Group as a reference for populating a freshly created
Connector.
+This is intended for Connectors that can translate an exported flow
definition, its parameter contexts, referenced assets, and any attached
component state into their own managed flow.
+The source Versioned Process Group is not installed onto the Connector;
instead, the Connector reads the source as input and updates its own flow to
mirror the source's configuration and state.
+
+Migration support is optional.
+Connectors opt in by implementing the optional capability interface
`org.apache.nifi.components.connector.migration.MigratableConnector`.
+A Connector that does not implement `MigratableConnector` is never offered as
a migration target and continues to behave as before.
+
+=== Migration contract
+
+`MigratableConnector` is a separate interface that the Connector class must
implement (typically alongside extending `AbstractConnector`). It exposes three
methods:
+
+- `isMigrationSupported(ConnectorMigrationContext)` is a cheap,
side-effect-free predicate.
+ It should inspect `context.getSourceFlow()` and return `true` only when the
Connector can be migrated from the source structure.
+ Implementations should limit themselves to structural checks such as
processor types, parameter names, and exported metadata.
+ Component state and asset content are not available at listing time and must
not be relied on; per-state precondition checks belong inside
`migrateConfiguration` or `migrateState`.
+ This method must not call
`ConnectorMigrationContext.copyAssetFromSource(...)` or any of the write APIs
described below; the framework wraps the eligibility-time context so any such
attempt is rejected with `IllegalStateException` and the Connector is filtered
out of the listing.
+- `migrateConfiguration(ConnectorMigrationContext)` records the configuration
the Connector wants on the other side of the migration.
+ The Connector reads `context.getSourceFlow()` and records configuration
changes via `context.setProperties(stepName, propertyValues)` (merge) or
`context.replaceProperties(stepName, propertyValues)` (full replacement of the
step).
+ These calls are applied to a working copy of the Connector's configuration
that the framework seeded with the Connector's current active configuration, so
the working copy always reflects the fully-merged result.
+ After `migrateConfiguration` returns, the framework drives
`Connector.applyUpdate(workingContext, activeContext)` from that merged
configuration so the managed Process Group is rebuilt by the same code path the
framework uses on restart.
+ The Connector must not call `updateFlow(...)` from this method; the
framework hands the Connector a read-only wrapper around the active
`FlowContext`, so any direct `updateFlow(...)` attempt is rejected with
`IllegalStateException`.
+ Component-state writes are also rejected in this phase:
`setComponentState(...)` throws because the managed components do not exist yet.
+- `migrateState(ConnectorMigrationContext)` runs after the framework has
rebuilt the managed Process Group from the merged configuration.
+ At this point every Processor and Controller Service in the source flow has
a corresponding managed component in the active `FlowContext`, so the Connector
can match each source `VersionedConfigurableExtension` to its managed
counterpart by versioned identifier.
+ The Connector records the desired `StateManager` state for each managed
component via `context.setComponentState(versionedComponentId,
versionedComponentState)`.
+ After `migrateState` returns, the framework writes each staged
`VersionedComponentState` into the corresponding managed component's live
`StateManager`.
+ Configuration writes (`setProperties`, `replaceProperties`) and direct
`updateFlow(...)` calls are rejected in this phase for the same reasons
described above.
+
+The two phases together form a single, all-or-nothing migration. The framework
only commits the merged configuration onto the Connector's persisted active
configuration after both `migrateState` and the framework's subsequent
state-application step have succeeded. Any failure in either phase rolls the
migration back: the managed Process Group is restored from `getInitialFlow()`,
the working configuration copy is discarded, any assets copied via
`copyAssetFromSource(...)` are deleted, and the Connector's persisted
configuration remains unchanged.
+
+=== Accessing the source flow
+
+`ConnectorMigrationContext` exposes the source `VersionedExternalFlow` using
`getSourceFlow()`.
+The source includes:
+
+- The exported `VersionedProcessGroup`
+- Parameter contexts
+- Referenced assets
+- Attached `VersionedComponentState` for `@Stateful` components
+- Export metadata such as flow name, registry identifiers, and version
+
+The same context also indicates whether the request came from a local Process
Group on the same NiFi instance or from an uploaded payload using
`isLocalMigration()`.
+
+The active `FlowContext` is available via `context.getActiveFlowContext()` and
is read-only during migration: it forwards reads (configuration, managed
Process Group) to the live active flow context but rejects `updateFlow(...)`
with `IllegalStateException`. The Connector should treat this object as a
window onto the current state of the Connector, not as a write target.
+
+=== Recording configuration changes
+
+A typical `migrateConfiguration` implementation translates the source flow
into a representation the Connector can rebuild from later (often a serialized
version of the source `VersionedExternalFlow` or a derived configuration), then
stages it onto the Connector's own configuration:
+
+[source,java]
+----
+@Override
+public void migrateConfiguration(final ConnectorMigrationContext context)
throws FlowUpdateException {
+ final VersionedExternalFlow sourceFlow = context.getSourceFlow();
+ final String serializedSourceFlow =
OBJECT_MAPPER.writeValueAsString(sourceFlow);
+
+ // Record the configuration the Connector wants on the other side of the
migration. The framework merges this
+ // delta onto the active configuration and then drives
Connector.applyUpdate(workingContext, activeContext),
+ // which is the same path used on every restart - so the managed Process
Group is rebuilt the same way after
+ // migration and after every subsequent restart.
+ context.setProperties(MIGRATION_STATE_STEP_NAME,
Map.of(MIGRATED_SOURCE_FLOW_PROPERTY_NAME, serializedSourceFlow));
+}
+----
+
+`setProperties` is additive: existing properties on the step keep their values
unless the staged map overrides them.
+`replaceProperties` is destructive: the step is reset to exactly the supplied
property map, and any pre-existing properties not present in the map are
removed.
+
+The Connector's `applyUpdate(workingContext, activeContext)` implementation
reads the merged configuration from the supplied `workingContext`, rebuilds the
desired managed Process Group, and calls
`getInitializationContext().updateFlow(activeContext, rebuiltFlow)` to install
it. Because the framework drives the same `applyUpdate(...)` call during
restart via `inheritConfiguration(...)`, the migrated managed flow is rebuilt
automatically on every restart from the persisted configuration; the Connector
does not need to do anything extra to make migrations survive restarts.
+
+=== Recording component state
+
+After the managed Process Group has been rebuilt, `migrateState` records the
`StateManager` state for individual managed components:
+
+[source,java]
+----
+@Override
+public void migrateState(final ConnectorMigrationContext context) {
+ final VersionedProcessGroup sourceGroup =
context.getSourceFlow().getFlowContents();
+ for (final VersionedProcessor sourceProcessor :
sourceGroup.getProcessors()) {
+ final VersionedComponentState sourceState =
sourceProcessor.getComponentState();
+ if (sourceState != null) {
+ context.setComponentState(sourceProcessor.getIdentifier(),
sourceState);
+ }
+ }
+}
+----
+
+The first argument to `setComponentState` is the source-side versioned
identifier; the framework resolves it to the matching managed Processor or
Controller Service and writes the supplied `VersionedComponentState` into that
component's live `StateManager`. Cluster-scope state is applied unconditionally
on every node; per-node local-scope state is selected by node ordinal so a
clustered migration installs the right entry on each node. Components whose
`@Stateful` annotation does not declare a requested scope are skipped with a
warning rather than failing the migration.
+
+`setComponentState` only stages the desired state; the framework writes it
after `migrateState` returns. If a Connector calls `setComponentState` more
than once for the same versioned identifier, only the last call is applied.
+
+=== Rewriting parameters and assets
+
+The exported source flow contains parameter contexts and any asset references
that were attached to exported parameters.
+Connectors are responsible for mapping those values into their own target
shape inside `migrateConfiguration`.
+
+A Connector can copy an asset from a local source Process Group into its own
asset namespace using `ConnectorMigrationContext.copyAssetFromSource(String)`.
+This method is available only for local migrations.
+It throws `IllegalArgumentException` when the supplied source asset identifier
is null or blank, and `IllegalStateException` when invoked for an
uploaded-payload migration.
+Uploaded payloads do not have access to a live source asset namespace, so
asset references from uploaded payloads must be handled outside of the
migration request.
+
Review Comment:
[Sonnet 5] Added a code example for asset migration to the developer guide,
based on the pattern used by MigrationTargetConnector in the system test
extensions.
##########
nifi-docs/src/main/asciidoc/developer-guide.adoc:
##########
@@ -2697,6 +2697,126 @@ The following command is used to generate a standard
binary distribution of Apac
`mvn clean install -Pcontrib-check`
+== Supporting migration from a Versioned Process Group
+
+Connectors can opt in to a one-time migration flow that uses a source
Versioned Process Group as a reference for populating a freshly created
Connector.
+This is intended for Connectors that can translate an exported flow
definition, its parameter contexts, referenced assets, and any attached
component state into their own managed flow.
+The source Versioned Process Group is not installed onto the Connector;
instead, the Connector reads the source as input and updates its own flow to
mirror the source's configuration and state.
+
+Migration support is optional.
+Connectors opt in by implementing the optional capability interface
`org.apache.nifi.components.connector.migration.MigratableConnector`.
+A Connector that does not implement `MigratableConnector` is never offered as
a migration target and continues to behave as before.
+
+=== Migration contract
+
+`MigratableConnector` is a separate interface that the Connector class must
implement (typically alongside extending `AbstractConnector`). It exposes three
methods:
+
+- `isMigrationSupported(ConnectorMigrationContext)` is a cheap,
side-effect-free predicate.
+ It should inspect `context.getSourceFlow()` and return `true` only when the
Connector can be migrated from the source structure.
+ Implementations should limit themselves to structural checks such as
processor types, parameter names, and exported metadata.
+ Component state and asset content are not available at listing time and must
not be relied on; per-state precondition checks belong inside
`migrateConfiguration` or `migrateState`.
+ This method must not call
`ConnectorMigrationContext.copyAssetFromSource(...)` or any of the write APIs
described below; the framework wraps the eligibility-time context so any such
attempt is rejected with `IllegalStateException` and the Connector is filtered
out of the listing.
+- `migrateConfiguration(ConnectorMigrationContext)` records the configuration
the Connector wants on the other side of the migration.
+ The Connector reads `context.getSourceFlow()` and records configuration
changes via `context.setProperties(stepName, propertyValues)` (merge) or
`context.replaceProperties(stepName, propertyValues)` (full replacement of the
step).
+ These calls are applied to a working copy of the Connector's configuration
that the framework seeded with the Connector's current active configuration, so
the working copy always reflects the fully-merged result.
+ After `migrateConfiguration` returns, the framework drives
`Connector.applyUpdate(workingContext, activeContext)` from that merged
configuration so the managed Process Group is rebuilt by the same code path the
framework uses on restart.
+ The Connector must not call `updateFlow(...)` from this method; the
framework hands the Connector a read-only wrapper around the active
`FlowContext`, so any direct `updateFlow(...)` attempt is rejected with
`IllegalStateException`.
+ Component-state writes are also rejected in this phase:
`setComponentState(...)` throws because the managed components do not exist yet.
+- `migrateState(ConnectorMigrationContext)` runs after the framework has
rebuilt the managed Process Group from the merged configuration.
+ At this point every Processor and Controller Service in the source flow has
a corresponding managed component in the active `FlowContext`, so the Connector
can match each source `VersionedConfigurableExtension` to its managed
counterpart by versioned identifier.
+ The Connector records the desired `StateManager` state for each managed
component via `context.setComponentState(versionedComponentId,
versionedComponentState)`.
+ After `migrateState` returns, the framework writes each staged
`VersionedComponentState` into the corresponding managed component's live
`StateManager`.
+ Configuration writes (`setProperties`, `replaceProperties`) and direct
`updateFlow(...)` calls are rejected in this phase for the same reasons
described above.
+
+The two phases together form a single, all-or-nothing migration. The framework
only commits the merged configuration onto the Connector's persisted active
configuration after both `migrateState` and the framework's subsequent
state-application step have succeeded. Any failure in either phase rolls the
migration back: the managed Process Group is restored from `getInitialFlow()`,
the working configuration copy is discarded, any assets copied via
`copyAssetFromSource(...)` are deleted, and the Connector's persisted
configuration remains unchanged.
+
+=== Accessing the source flow
+
+`ConnectorMigrationContext` exposes the source `VersionedExternalFlow` using
`getSourceFlow()`.
+The source includes:
+
+- The exported `VersionedProcessGroup`
+- Parameter contexts
+- Referenced assets
+- Attached `VersionedComponentState` for `@Stateful` components
+- Export metadata such as flow name, registry identifiers, and version
+
+The same context also indicates whether the request came from a local Process
Group on the same NiFi instance or from an uploaded payload using
`isLocalMigration()`.
+
+The active `FlowContext` is available via `context.getActiveFlowContext()` and
is read-only during migration: it forwards reads (configuration, managed
Process Group) to the live active flow context but rejects `updateFlow(...)`
with `IllegalStateException`. The Connector should treat this object as a
window onto the current state of the Connector, not as a write target.
+
+=== Recording configuration changes
+
+A typical `migrateConfiguration` implementation translates the source flow
into a representation the Connector can rebuild from later (often a serialized
version of the source `VersionedExternalFlow` or a derived configuration), then
stages it onto the Connector's own configuration:
+
+[source,java]
+----
+@Override
+public void migrateConfiguration(final ConnectorMigrationContext context)
throws FlowUpdateException {
+ final VersionedExternalFlow sourceFlow = context.getSourceFlow();
+ final String serializedSourceFlow =
OBJECT_MAPPER.writeValueAsString(sourceFlow);
+
+ // Record the configuration the Connector wants on the other side of the
migration. The framework merges this
+ // delta onto the active configuration and then drives
Connector.applyUpdate(workingContext, activeContext),
+ // which is the same path used on every restart - so the managed Process
Group is rebuilt the same way after
+ // migration and after every subsequent restart.
+ context.setProperties(MIGRATION_STATE_STEP_NAME,
Map.of(MIGRATED_SOURCE_FLOW_PROPERTY_NAME, serializedSourceFlow));
+}
+----
+
+`setProperties` is additive: existing properties on the step keep their values
unless the staged map overrides them.
+`replaceProperties` is destructive: the step is reset to exactly the supplied
property map, and any pre-existing properties not present in the map are
removed.
+
+The Connector's `applyUpdate(workingContext, activeContext)` implementation
reads the merged configuration from the supplied `workingContext`, rebuilds the
desired managed Process Group, and calls
`getInitializationContext().updateFlow(activeContext, rebuiltFlow)` to install
it. Because the framework drives the same `applyUpdate(...)` call during
restart via `inheritConfiguration(...)`, the migrated managed flow is rebuilt
automatically on every restart from the persisted configuration; the Connector
does not need to do anything extra to make migrations survive restarts.
+
+=== Recording component state
+
+After the managed Process Group has been rebuilt, `migrateState` records the
`StateManager` state for individual managed components:
+
+[source,java]
+----
+@Override
+public void migrateState(final ConnectorMigrationContext context) {
+ final VersionedProcessGroup sourceGroup =
context.getSourceFlow().getFlowContents();
+ for (final VersionedProcessor sourceProcessor :
sourceGroup.getProcessors()) {
+ final VersionedComponentState sourceState =
sourceProcessor.getComponentState();
+ if (sourceState != null) {
+ context.setComponentState(sourceProcessor.getIdentifier(),
sourceState);
+ }
+ }
+}
+----
+
+The first argument to `setComponentState` is the source-side versioned
identifier; the framework resolves it to the matching managed Processor or
Controller Service and writes the supplied `VersionedComponentState` into that
component's live `StateManager`. Cluster-scope state is applied unconditionally
on every node; per-node local-scope state is selected by node ordinal so a
clustered migration installs the right entry on each node. Components whose
`@Stateful` annotation does not declare a requested scope are skipped with a
warning rather than failing the migration.
+
+`setComponentState` only stages the desired state; the framework writes it
after `migrateState` returns. If a Connector calls `setComponentState` more
than once for the same versioned identifier, only the last call is applied.
+
+=== Rewriting parameters and assets
+
+The exported source flow contains parameter contexts and any asset references
that were attached to exported parameters.
+Connectors are responsible for mapping those values into their own target
shape inside `migrateConfiguration`.
+
+A Connector can copy an asset from a local source Process Group into its own
asset namespace using `ConnectorMigrationContext.copyAssetFromSource(String)`.
+This method is available only for local migrations.
+It throws `IllegalArgumentException` when the supplied source asset identifier
is null or blank, and `IllegalStateException` when invoked for an
uploaded-payload migration.
+Uploaded payloads do not have access to a live source asset namespace, so
asset references from uploaded payloads must be handled outside of the
migration request.
+
+=== Sensitive values
+
+Sensitive parameter values and other secret material are not present in the
exported source flow.
+Connectors must leave those values unset during migration.
+After the migration completes, the user supplies the missing values through
the normal Configure step before starting the Connector.
+
+=== Additional expectations
+
+- `isMigrationSupported(...)` must not mutate flow state, parameter values, or
assets.
+- `migrateConfiguration` and `migrateState` should either complete
successfully or throw an exception. Any thrown exception aborts the migration
and rolls back every side effect: copied assets are deleted, the managed
Process Group is restored from `getInitialFlow()`, every component's
`StateManager` state is cleared, and the persisted active configuration is left
unchanged.
+- Migration is rejected when the Connector's currently managed flow no longer
matches the flow returned by its `getInitialFlow()`.
+ The framework compares the two structurally (ignoring environmental
differences such as component scheduled state) at the time
`migrateConfiguration` is invoked, and again whenever the `MIGRATE` allowable
action is evaluated, so the user is shown why migration is unavailable.
+ This check is restart-safe because both inputs are recomputed each time from
durable sources: the Connector's persisted configuration drives
`getInitialFlow()`, and the current managed flow is read from the live flow
tree restored from `flow.json.gz`.
+- Migration is one-shot from the framework's point of view.
+ Once the Connector's flow has been updated by a migration, its currently
managed flow no longer matches `getInitialFlow()`, and a subsequent migration
request will be rejected until the Connector is discarded and a fresh one is
created.
Review Comment:
[Sonnet 5] No — this text was stale relative to the eligibility check added
in the July 16 commit, which now uses Connector.isModified() (configuration
deviates from defaults, or any managed component has stored state) rather than
a structural comparison against getInitialFlow(). Reworded this section to
describe the current check.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorMigrationManager.java:
##########
@@ -0,0 +1,832 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.components.connector;
+
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.components.ConfigurableComponent;
+import org.apache.nifi.components.connector.migration.MigratableConnector;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.connectable.Connection;
+import org.apache.nifi.connectable.Port;
+import org.apache.nifi.controller.FlowController;
+import org.apache.nifi.controller.ProcessorNode;
+import org.apache.nifi.controller.ScheduledState;
+import org.apache.nifi.controller.flow.FlowManager;
+import org.apache.nifi.controller.queue.QueueSize;
+import org.apache.nifi.controller.service.ControllerServiceNode;
+import org.apache.nifi.controller.service.ControllerServiceState;
+import org.apache.nifi.flow.ExternalControllerServiceReference;
+import org.apache.nifi.flow.VersionedComponentState;
+import org.apache.nifi.flow.VersionedExternalFlow;
+import org.apache.nifi.flow.VersionedExternalFlowMetadata;
+import org.apache.nifi.flow.VersionedNodeState;
+import org.apache.nifi.flow.synchronization.VersionedComponentStateValidator;
+import org.apache.nifi.groups.ProcessGroup;
+import org.apache.nifi.nar.NarCloseable;
+import org.apache.nifi.registry.flow.RegisteredFlow;
+import org.apache.nifi.registry.flow.RegisteredFlowSnapshot;
+import org.apache.nifi.registry.flow.RegisteredFlowSnapshotMetadata;
+import org.apache.nifi.registry.flow.VersionControlInformation;
+import org.apache.nifi.registry.flow.VersionedFlowState;
+import org.apache.nifi.registry.flow.VersionedFlowStatus;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.BooleanSupplier;
+
+public class StandardConnectorMigrationManager implements
ConnectorMigrationManager {
+ private static final Logger logger =
LoggerFactory.getLogger(StandardConnectorMigrationManager.class);
+
+ private static final String MIGRATED_SOURCE_PREFIX = "(Migrated) ";
+
+ private final FlowController flowController;
+ private final ConnectorFlowSnapshotProvider snapshotProvider;
+
+ public StandardConnectorMigrationManager(final FlowController
flowController, final ConnectorFlowSnapshotProvider snapshotProvider) {
+ this.flowController = Objects.requireNonNull(flowController,
"FlowController is required");
+ this.snapshotProvider = Objects.requireNonNull(snapshotProvider,
"ConnectorFlowSnapshotProvider is required");
+ }
+
+ @Override
+ public List<ConnectorMigrationSource> listMigrationSources(final String
connectorId) {
+ final ConnectorNode connector = getRequiredConnector(connectorId);
+ final List<ConnectorMigrationSource> migrationSources = new
ArrayList<>();
+
+ for (final ProcessGroup processGroup : getCandidateSourceGroups()) {
+ buildMigrationSourceForListing(connector,
processGroup).ifPresent(migrationSources::add);
+ }
+
+ return migrationSources;
+ }
+
+ /**
+ * Builds a {@link ConnectorMigrationSource} for the given Process Group
when it is structurally compatible with the
+ * target Connector. Returns {@link Optional#empty()} when the Process
Group fails a hard filter (not under version
+ * control, already managed by a Connector, or rejected by the Connector's
own structural compatibility check).
+ * Returns a populated source — possibly with {@link
ConnectorMigrationSource#isReadyForMigration()} set to
+ * {@code false} and an ineligibility reason — when the group is
compatible but currently in a state that requires
+ * user remediation before migration can proceed.
+ */
+ private Optional<ConnectorMigrationSource>
buildMigrationSourceForListing(final ConnectorNode connector, final
ProcessGroup processGroup) {
+ final VersionControlInformation versionControlInformation =
processGroup.getVersionControlInformation();
+ if (versionControlInformation == null) {
+ return Optional.empty();
+ }
+ if (processGroup.getConnectorIdentifier().isPresent()) {
+ return Optional.empty();
+ }
+
+ final VersionedExternalFlow sourceFlow =
obtainSourceFlowForListing(processGroup);
+ if (!isConnectorMigrationSupported(connector, sourceFlow)) {
+ return Optional.empty();
+ }
+
+ final List<String> stateIneligibilityReasons =
getStateIneligibilityReasons(processGroup, versionControlInformation,
sourceFlow);
+ return Optional.of(toMigrationSource(processGroup,
versionControlInformation, stateIneligibilityReasons));
+ }
+
+ @Override
+ public void verifyEligibility(final String connectorId, final String
processGroupId) {
+ final ConnectorNode connector = getRequiredConnector(connectorId);
+ final ProcessGroup processGroup =
getRequiredSourceProcessGroup(processGroupId);
+ final String ineligibilityReason = getIneligibilityReason(connector,
processGroup);
+ if (ineligibilityReason != null) {
+ throw new IllegalStateException(ineligibilityReason);
+ }
+ }
+
+ @Override
+ public void verifyConnectorReadyForMigration(final String connectorId) {
+ final ConnectorNode connector = getRequiredConnector(connectorId);
+ verifyConnectorCanReceiveMigration(connector);
+ verifyTargetIsAtInitialFlow(connector);
+ flowController.getConnectorRepository().verifyMigration(connectorId);
+ }
+
+ @Override
+ public void migrateFromVersionedFlow(final String connectorId, final
String processGroupId, final VersionedExternalFlow sourceFlow,
+ final BooleanSupplier cancellationCheck) throws
FlowUpdateException {
+ // Mark the migration in progress so a concurrent flow synchronization
does not reclaim the assets this
+ // migration copies before the managed Process Group is rebuilt to
reference them. The marker is cleared in
+ // the finally block whether the migration succeeds, fails, or rolls
back.
+ flowController.getConnectorRepository().beginMigration(connectorId);
+ try {
+ performMigration(connectorId, processGroupId, sourceFlow,
cancellationCheck);
+ } finally {
+ flowController.getConnectorRepository().endMigration(connectorId);
+ }
+ }
+
+ private void performMigration(final String connectorId, final String
processGroupId, final VersionedExternalFlow sourceFlow,
+ final BooleanSupplier cancellationCheck) throws
FlowUpdateException {
+ final BooleanSupplier cancellation = cancellationCheck == null ? () ->
false : cancellationCheck;
+ final ConnectorNode connector = getRequiredConnector(connectorId);
+ verifyConnectorCanReceiveMigration(connector);
+ verifyTargetIsAtInitialFlow(connector);
+ // Give the ConnectorConfigurationProvider the chance to reject the
migration before any flow manipulation
+ // begins, so a rejection aborts the migration without exercising the
rollback machinery.
+ flowController.getConnectorRepository().verifyMigration(connectorId);
+ validateMigrationSource(sourceFlow);
+
+ final boolean localMigration = processGroupId != null;
+ final ProcessGroup sourceProcessGroup = localMigration ?
getRequiredSourceProcessGroup(processGroupId) : null;
+ // Seed the migration context with a working clone of the connector's
active configuration. The connector's
+ // setProperties/replaceProperties calls during
migrateConfiguration(...) mutate this clone directly, so it
+ // always holds the fully-merged result. The active configuration is
left untouched until commit.
+ final MutableConnectorConfigurationContext workingConfiguration =
connector.getActiveFlowContext().getConfigurationContext().clone();
+ final StandardFrameworkConnectorMigrationContext migrationContext =
new StandardFrameworkConnectorMigrationContext(
+ connectorId,
+ sourceFlow,
+ localMigration,
+ connector.getActiveFlowContext(),
+ workingConfiguration,
+ flowController.getAssetManager(),
+ flowController.getConnectorRepository(),
+ flowController.getStateManagerProvider(),
+ flowController
+ );
+
+ final Connector rawConnector = connector.getConnector();
+ if (!(rawConnector instanceof final MigratableConnector
migratableConnector)) {
+ throw new FlowUpdateException("Connector " + connectorId + " does
not support migration from the provided source flow.");
Review Comment:
[Sonnet 5] Updated as suggested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
