bobpaulin commented on code in PR #11240:
URL: https://github.com/apache/nifi/pull/11240#discussion_r3597946897
##########
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.
Review Comment:
```suggestion
- The target Connector must be not yet configured, stopped, and not
previously migrated.
```
How recent the connector was created does not seem to come into play. It's
whether it was configured prior to the attempt to migrate.
##########
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:
```suggestion
. Open the Connector configuration and supply any missing secrets.
```
Since only Connector Secrets are supported I'm not sure mentioning sensitive
values make sense unless referring to the source flow.
##########
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:
It is mentioned later that the applyUpdate will be performed to update the
active context but it might be worth mentioning here 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.
+
+=== 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:
Is the point about getInitialFlow still true with the recent changes?
##########
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.");
+ }
+
+ try (final NarCloseable ignored =
NarCloseable.withComponentNarLoader(flowController.getExtensionManager(),
rawConnector.getClass(), connectorId)) {
+ if (!migratableConnector.isMigrationSupported(migrationContext)) {
+ throw new FlowUpdateException("Connector " + connectorId + "
does not support migration from the provided source flow.");
+ }
+ }
+
+ // Phase 1: drive migrateConfiguration(...) ->
applyMigratedConfiguration(...). The connector records the
+ // configuration it wants on the other side of migration by mutating
the working configuration clone seeded
+ // into the migration context; the framework then drives
Connector.applyUpdate(workingContext, activeContext)
+ // so the managed Process Group is rebuilt from that merged
configuration. The active configuration itself is
+ // intentionally not written here: the merged configuration is held
until the state phase below succeeds, then
+ // committed by commitMigratedConfiguration(...). That ordering keeps
rollback simple - any failure in either
+ // phase can be recovered by restoring the initial flow because the
active configuration was never mutated.
+ try (final NarCloseable ignored =
NarCloseable.withComponentNarLoader(flowController.getExtensionManager(),
rawConnector.getClass(), connectorId)) {
+
flowController.getConnectorRepository().syncAssetsFromProvider(connector);
+ migratableConnector.migrateConfiguration(migrationContext);
+
+ if (cancellation.getAsBoolean()) {
+ throw new FlowUpdateException("Migration of Connector " +
connectorId + " was cancelled after migrateConfiguration() completed.");
+ }
+ } catch (final FlowUpdateException e) {
+ failMigration(connector, migrationContext);
+ throw e;
+ } catch (final Exception e) {
+ failMigration(connector, migrationContext);
+ throw new FlowUpdateException("Failed to migrate Connector " +
connectorId + " from the provided source flow", e);
+ }
+
+ final ConnectorConfiguration mergedConfiguration;
+ try {
+ mergedConfiguration =
connector.applyMigratedConfiguration(migrationContext.getMergedConfiguration());
+ } catch (final FlowUpdateException e) {
+ failMigration(connector, migrationContext);
+ throw e;
+ } catch (final Exception e) {
+ failMigration(connector, migrationContext);
+ throw new FlowUpdateException("Failed to apply migrated
configuration for Connector " + connectorId, e);
+ }
+
+ // Phase 2: drive migrateState(...) -> drain staged component-state
writes -> apply each entry to the live
+ // StateManager. The managed Process Group has been rebuilt by
applyMigratedConfiguration(...) above, so the
+ // managed components exist before the connector tries to record state
for them. Wrong-phase calls
+ // (setProperties/replaceProperties) now throw IllegalStateException
from the context; updateFlow(...) on
+ // the active flow context still throws via the MigrationFlowContext
wrapper.
+
migrationContext.setPhase(StandardFrameworkConnectorMigrationContext.Phase.STATE);
+ try (final NarCloseable ignored =
NarCloseable.withComponentNarLoader(flowController.getExtensionManager(),
rawConnector.getClass(), connectorId)) {
+ migratableConnector.migrateState(migrationContext);
+
+ if (cancellation.getAsBoolean()) {
+ throw new FlowUpdateException("Migration of Connector " +
connectorId + " was cancelled after migrateState() completed.");
+ }
+ } catch (final FlowUpdateException e) {
+ failMigration(connector, migrationContext);
+ throw e;
+ } catch (final Exception e) {
+ failMigration(connector, migrationContext);
+ throw new FlowUpdateException("Failed to migrate state for
Connector " + connectorId, e);
+ }
+
+ final Map<String, VersionedComponentState> stagedStates =
migrationContext.drainStagedComponentStates();
+ try {
+ applyMigratedComponentStates(connector, stagedStates);
+ } catch (final Exception e) {
+ failMigration(connector, migrationContext);
+ throw new FlowUpdateException("Failed to apply migrated component
state for Connector " + connectorId, e);
+ }
+
+ // Commit: both phases have succeeded, so persist the merged
configuration onto the active configuration.
+ // From this point on the migration is durable across restarts:
inheritConfiguration(...) on the next load will
+ // rebuild the managed Process Group from the committed configuration
via the same Connector.applyUpdate(...)
+ // path used here.
+ try {
+ connector.commitMigratedConfiguration(mergedConfiguration);
+ } catch (final Exception e) {
+ failMigration(connector, migrationContext);
+ throw new FlowUpdateException("Failed to commit migrated
configuration for Connector " + connectorId, e);
+ }
+
+
migrationContext.setPhase(StandardFrameworkConnectorMigrationContext.Phase.COMPLETED);
+
+ // The migration is durable now that the merged configuration is
committed, so notify the
+ // ConnectorConfigurationProvider that the migration completed. The
source Process Group identifier is null for
+ // an uploaded-payload migration. A failure here is logged rather than
rolled back, because rolling back a
+ // committed migration would silently resurrect it on the next restart.
+ try {
+
flowController.getConnectorRepository().notifyMigrationComplete(connectorId,
processGroupId);
+ } catch (final Exception e) {
+ logger.warn("Connector {} migration completed and is durable, but
notifying the configuration provider of completion failed", connectorId, e);
+ }
+
+ // Finalize: the migration is already durable at this point because
the merged configuration has been committed
+ // onto the active configuration. The remaining work is cosmetic -
disabling and renaming the source Process
+ // Group so the operator can tell it apart from the new managed flow.
Failures (or a late cancellation) below
+ // are logged but no longer trigger a rollback because rolling back a
committed migration would leave the
+ // active configuration migrated while the managed Process Group was
reset to the initial flow, which would
+ // silently resurrect the migration on the next restart.
+ if (cancellation.getAsBoolean()) {
+ logger.warn("Migration of Connector {} was cancelled after the
merged configuration was committed; the migration is durable but the source
Process Group was not finalized", connectorId);
+ return;
+ }
+
+ if (sourceProcessGroup != null) {
+ try {
+ disableAndRenameSourceProcessGroup(sourceProcessGroup);
+ } catch (final Exception e) {
+ logger.warn("Failed to finalize source Process Group {} after
a successful migration of Connector {}; manual cleanup may be required",
+ sourceProcessGroup.getIdentifier(), connectorId, e);
+ }
+ }
+ }
+
+ /**
+ * Applies the staged component-state writes drained from {@code
migrateState(...)} onto the live components in
+ * the connector's managed Process Group. For each entry the framework:
+ * <ol>
+ * <li>resolves the managed component by its versioned identifier;</li>
+ * <li>looks up the component's runtime instance identifier and
fetches its {@link StateManager} via
+ * {@link
org.apache.nifi.components.state.StateManagerProvider#getStateManager(String)
+ * StateManagerProvider.getStateManager(...)};</li>
+ * <li>writes the cluster-scope state, when the component declares
{@link Scope#CLUSTER} on its
+ * {@code @Stateful} annotation;</li>
+ * <li>writes the local-scope state for this node, when the component
declares {@link Scope#LOCAL}. The
+ * local-state entry is picked by node ordinal so a clustered
migration installs the right per-node state
+ * on each node, matching the framework's normal versioned-flow
synchronizer.</li>
+ * </ol>
+ * Skips components whose {@code @Stateful} annotation does not declare
the requested scope and logs a warning,
+ * matching the behavior of {@code
StandardVersionedComponentSynchronizer.restoreComponentState(...)}.
+ *
+ * <p>
+ * This loop applies entries sequentially and does not roll back partial
writes on its own. The rollback path that
+ * runs when any entry throws (see {@link #clearConnectorComponentState})
wipes every managed component's
+ * StateManager state, so a partially-applied set of writes is cleared
along with the rest of the migration's
+ * side effects rather than left behind.
+ * </p>
+ */
+ private void applyMigratedComponentStates(final ConnectorNode connector,
final Map<String, VersionedComponentState> stagedStates) throws IOException {
+ if (stagedStates.isEmpty()) {
+ return;
+ }
+
+ final FrameworkFlowContext activeFlowContext =
connector.getActiveFlowContext();
+ if (activeFlowContext == null) {
+ throw new IllegalStateException("Cannot apply migrated component
state for Connector " + connector.getIdentifier()
+ + " because it has no active flow context.");
+ }
+ final ProcessGroup managedGroup =
activeFlowContext.getManagedProcessGroup();
+ if (managedGroup == null) {
+ throw new IllegalStateException("Cannot apply migrated component
state for Connector " + connector.getIdentifier()
+ + " because its managed Process Group does not exist.");
+ }
+
+ final Map<String, ProcessorNode> processorsByVersionedId = new
HashMap<>();
+ for (final ProcessorNode processor : managedGroup.findAllProcessors())
{
+ processor.getVersionedComponentId().ifPresent(versionedId ->
processorsByVersionedId.put(versionedId, processor));
+ }
+
+ final Map<String, ControllerServiceNode> servicesByVersionedId = new
HashMap<>();
+ for (final ControllerServiceNode service :
managedGroup.findAllControllerServices()) {
+ service.getVersionedComponentId().ifPresent(versionedId ->
servicesByVersionedId.put(versionedId, service));
+ }
+
+ final int localNodeOrdinal = flowController.getLocalNodeOrdinal();
+
+ for (final Map.Entry<String, VersionedComponentState> entry :
stagedStates.entrySet()) {
+ final String versionedId = entry.getKey();
+ final VersionedComponentState desiredState = entry.getValue();
+
+ final ProcessorNode processor =
processorsByVersionedId.get(versionedId);
+ if (processor != null) {
+ applyComponentState(processor.getIdentifier(),
processor.getComponent(), desiredState, localNodeOrdinal);
+ continue;
+ }
+ final ControllerServiceNode service =
servicesByVersionedId.get(versionedId);
+ if (service != null) {
+ applyComponentState(service.getIdentifier(),
service.getControllerServiceImplementation(), desiredState, localNodeOrdinal);
+ continue;
+ }
+ logger.warn("Connector {} migrateState() requested state for
component [{}] but no managed Processor or Controller Service with that
versioned identifier exists; skipping",
+ connector.getIdentifier(), versionedId);
+ }
+ }
+
+ private void failMigration(final ConnectorNode connector, final
StandardFrameworkConnectorMigrationContext migrationContext) {
+
migrationContext.setPhase(StandardFrameworkConnectorMigrationContext.Phase.COMPLETED);
+ rollbackMigration(connector, migrationContext);
+ }
+
+ private void applyComponentState(final String runtimeComponentId, final
ConfigurableComponent component, final VersionedComponentState desiredState,
final int localNodeOrdinal)
+ throws IOException {
+ final StateManager stateManager =
flowController.getStateManagerProvider().getStateManager(runtimeComponentId);
+ if (stateManager == null) {
+ logger.warn("No StateManager registered for migrated component
[{}]; skipping state write", runtimeComponentId);
+ return;
+ }
+
+ final Set<Scope> declaredScopes = declaredStatefulScopes(component);
+
+ final Map<String, String> clusterState =
desiredState.getClusterState();
+ if (clusterState != null && !clusterState.isEmpty()) {
+ if (declaredScopes.contains(Scope.CLUSTER)) {
+ stateManager.setState(clusterState, Scope.CLUSTER);
+ } else {
+ logger.warn("Component [{}] does not declare @Stateful scope
CLUSTER; skipping cluster state migration", runtimeComponentId);
+ }
+ }
+
+ final List<VersionedNodeState> localNodeStates =
desiredState.getLocalNodeStates();
+ if (localNodeStates != null && !localNodeStates.isEmpty()) {
+ if (!declaredScopes.contains(Scope.LOCAL)) {
+ logger.warn("Component [{}] does not declare @Stateful scope
LOCAL; skipping local state migration", runtimeComponentId);
+ return;
+ }
+ if (localNodeOrdinal < 0) {
+ logger.warn("Local node ordinal is unknown; skipping local
state migration for component [{}]", runtimeComponentId);
+ return;
+ }
+ if (localNodeOrdinal >= localNodeStates.size()) {
+ logger.warn("Component [{}] migration has {} local node state
entries but this node ordinal is {}; skipping local state migration",
+ runtimeComponentId, localNodeStates.size(),
localNodeOrdinal);
+ return;
+ }
+ final VersionedNodeState nodeState =
localNodeStates.get(localNodeOrdinal);
+ if (nodeState == null || nodeState.getState() == null ||
nodeState.getState().isEmpty()) {
+ return;
+ }
+ stateManager.setState(nodeState.getState(), Scope.LOCAL);
+ }
+ }
+
+ private static Set<Scope> declaredStatefulScopes(final
ConfigurableComponent component) {
+ if (component == null) {
+ return Set.of();
+ }
+ final Stateful stateful =
component.getClass().getAnnotation(Stateful.class);
+ if (stateful == null) {
+ return Set.of();
+ }
+ return EnumSet.copyOf(Arrays.asList(stateful.scopes()));
+ }
+
+ private void rollbackMigration(final ConnectorNode connector, final
StandardFrameworkConnectorMigrationContext migrationContext) {
+ clearConnectorComponentState(connector);
+
+ final Set<String> copiedAssetIds =
migrationContext.getCopiedAssetIds();
+ if (!copiedAssetIds.isEmpty()) {
+ try {
+
flowController.getConnectorRepository().deleteAssets(connector.getIdentifier(),
copiedAssetIds);
+ } catch (final Exception e) {
+ logger.warn("Failed to delete copied assets {} for Connector
{} during migration rollback", copiedAssetIds, connector.getIdentifier(), e);
+ }
+ }
+
+ try {
+ connector.loadInitialFlow();
+ } catch (final FlowUpdateException e) {
+ // Do not propagate the rollback failure: the caller's primary
need is to see why the migration originally
+ // failed, not how the recovery attempt failed. The operator can
read the rollback failure from the log.
+ logger.error("Failed to restore Connector {} to its initial state
after migration failure; manual cleanup may be required",
+ connector.getIdentifier(), e);
+ return;
+ }
+
+
flowController.getConnectorRepository().discardWorkingConfiguration(connector);
+ }
+
+ private void clearConnectorComponentState(final ConnectorNode connector) {
+ // Migration is only permitted on a Connector whose active flow has
not been modified from its initial flow.
+ // Rollback restores that initial flow immediately after this call, so
clearing state for every Processor and
+ // Controller Service currently in the managed flow is acceptable: any
component carried over from the initial
+ // flow is rebuilt fresh by loadInitialFlow(), and any component
introduced by the failed migration is
+ // discarded. Clearing the whole managed flow this way avoids the
brittle approach of trying to map source-flow
+ // versioned identifiers to runtime component identifiers.
+ final FrameworkFlowContext activeFlowContext =
connector.getActiveFlowContext();
+ if (activeFlowContext == null) {
+ return;
+ }
+
+ final ProcessGroup managedGroup =
activeFlowContext.getManagedProcessGroup();
+ if (managedGroup == null) {
+ return;
+ }
+
+ for (final ProcessorNode processor : managedGroup.findAllProcessors())
{
+ clearComponentState(processor.getIdentifier());
+ }
+ for (final ControllerServiceNode controllerService :
managedGroup.findAllControllerServices()) {
+ clearComponentState(controllerService.getIdentifier());
+ }
+ }
+
+ private void clearComponentState(final String componentIdentifier) {
+ final StateManager stateManager =
flowController.getStateManagerProvider().getStateManager(componentIdentifier);
+ if (stateManager == null) {
+ return;
+ }
+
+ try {
+ stateManager.clear(Scope.LOCAL);
+ } catch (final Exception e) {
+ logger.warn("Failed to clear LOCAL state for component {} during
migration rollback", componentIdentifier, e);
+ }
+
+ try {
+ stateManager.clear(Scope.CLUSTER);
+ } catch (final Exception e) {
+ logger.warn("Failed to clear CLUSTER state for component {} during
migration rollback", componentIdentifier, e);
+ }
+ }
+
+ private ConnectorNode getRequiredConnector(final String connectorId) {
+ final ConnectorNode connector =
flowController.getConnectorRepository().getConnector(connectorId,
ConnectorSyncMode.LOCAL_ONLY);
+ if (connector == null) {
+ throw new IllegalArgumentException("Could not find Connector with
ID " + connectorId);
+ }
+
+ return connector;
+ }
+
+ private ProcessGroup getRequiredSourceProcessGroup(final String
processGroupId) {
+ final ProcessGroup rootGroup =
flowController.getFlowManager().getRootGroup();
+ final ProcessGroup processGroup = rootGroup == null ? null :
rootGroup.findProcessGroup(processGroupId);
+ if (processGroup == null) {
+ throw new IllegalArgumentException("Could not find Process Group
with ID " + processGroupId);
+ }
+
+ return processGroup;
+ }
+
+ private List<ProcessGroup> getCandidateSourceGroups() {
+ final FlowManager flowManager = flowController.getFlowManager();
+ final ProcessGroup rootGroup = flowManager.getRootGroup();
+ if (rootGroup == null) {
+ return Collections.emptyList();
+ }
+
+ return new ArrayList<>(rootGroup.findAllProcessGroups());
+ }
+
+ private String getIneligibilityReason(final ConnectorNode connector, final
ProcessGroup processGroup) {
+ final VersionControlInformation versionControlInformation =
processGroup.getVersionControlInformation();
+ if (versionControlInformation == null) {
+ return "The Process Group is not under version control.";
+ }
+
+ if (processGroup.getConnectorIdentifier().isPresent()) {
+ return "The Process Group is managed by another Connector and
cannot be used as a migration source.";
+ }
+
+ final VersionedExternalFlow sourceFlow =
obtainSourceFlowForListing(processGroup);
+
+ final List<String> stateIneligibilityReasons =
getStateIneligibilityReasons(processGroup, versionControlInformation,
sourceFlow);
+ if (!stateIneligibilityReasons.isEmpty()) {
+ // Each entry describes a distinct condition the user can address.
Joining with newlines keeps the
+ // individual reasons identifiable (e.g. "There are 8 running
Processors.\nThere are 3 queued FlowFiles.")
+ // rather than collapsing them into a single run-on sentence when
they are surfaced to the operator.
+ return String.join(System.lineSeparator(),
stateIneligibilityReasons);
+ }
+
+ if (!isConnectorMigrationSupported(connector, sourceFlow)) {
+ return "The Connector does not support migration from this Process
Group.";
+ }
+
+ return null;
+ }
+
+ /**
+ * Reports every condition that currently prevents the Process Group from
being used as a migration source.
+ * Reasons describe conditions the user can remediate (stopping
processors, draining queues, refreshing version
+ * control, etc.) without changing the structural compatibility of the
source flow. The returned list contains
+ * one entry per applicable condition so the user can address them
together; an empty list means the Process
+ * Group is in a state that allows migration. Callers must ensure the
supplied {@code versionControlInformation}
+ * matches the {@code processGroup}; this method does not tolerate a null
value.
+ *
+ * @return ordered list of state-based ineligibility reasons, or an empty
list if the Process Group is in a state
+ * that allows migration
+ */
+ private List<String> getStateIneligibilityReasons(final ProcessGroup
processGroup, final VersionControlInformation versionControlInformation, final
VersionedExternalFlow sourceFlow) {
+ final List<String> reasons = new ArrayList<>();
+
+ final VersionedFlowStatus versionedFlowStatus =
versionControlInformation.getStatus();
+ final VersionedFlowState state = versionedFlowStatus == null ? null :
versionedFlowStatus.getState();
+ if (state != VersionedFlowState.UP_TO_DATE) {
+ reasons.add(describeIneligibleVersionedFlowState(state));
+ }
+
+ final int runningProcessorCount = countRunningProcessors(processGroup);
+ if (runningProcessorCount > 0) {
+ reasons.add("There " + pluralizeIsAre(runningProcessorCount) + " "
+ runningProcessorCount + " running or enabled "
+ + pluralizeNoun(runningProcessorCount, "processor",
"processors") + ".");
+ }
+
+ final int enabledServiceCount =
countEnabledControllerServices(processGroup);
+ if (enabledServiceCount > 0) {
+ reasons.add("There " + pluralizeIsAre(enabledServiceCount) + " " +
enabledServiceCount + " enabled controller "
+ + pluralizeNoun(enabledServiceCount, "service",
"services") + ".");
+ }
+
+ final long queuedFlowFileCount = countQueuedFlowFiles(processGroup);
+ if (queuedFlowFileCount > 0) {
+ reasons.add("There " + pluralizeIsAre(queuedFlowFileCount) + " " +
queuedFlowFileCount + " queued "
+ + pluralizeNoun(queuedFlowFileCount, "FlowFile",
"FlowFiles") + ".");
+ }
+
+ final int externalControllerServiceCount =
countExternalControllerServices(sourceFlow);
+ if (externalControllerServiceCount > 0) {
+ reasons.add("The Process Group references " +
externalControllerServiceCount + " controller "
+ + pluralizeNoun(externalControllerServiceCount, "service",
"services") + " from outside the Process Group.");
+ }
+
+ return reasons;
+ }
+
+ private static String pluralizeIsAre(final long count) {
+ return count == 1L ? "is" : "are";
+ }
+
+ private static String pluralizeNoun(final long count, final String
singular, final String plural) {
Review Comment:
While correct grammatically does this make log analysis harder?
##########
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:
May want to mention the exports should include state (if it exists).
##########
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:
```suggestion
Only Connectors that implement migration support can use this capability.
```
They opt in by implementing the support interfaces. While this is outlined
in the developer guide an administrator may confuse opt in for some type of
configuration rather than code.
##########
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:
```suggestion
Connectors can implement a one-time migration flow that uses a source
Versioned Process Group as a reference for populating a freshly created
Connector.
```
##########
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:
Code snippet for Asset Migration might be helpful. As asset needs to be
migrated and reference set.
```
final ConnectorMigrationContext context = ...
final VersionedParameter sourceParameter = ...
final String targetStepName = ...
final String targetPropertyName = ...
if (sourceParameter == null || sourceParameter.getReferencedAssets() ==
null
|| sourceParameter.getReferencedAssets().isEmpty() ||
!context.isLocalMigration()) {
return;
}
final Set<String> copiedAssetIds = new LinkedHashSet<>();
for (final VersionedAsset asset : sourceParameter.getReferencedAssets())
{
if (asset.getIdentifier() != null &&
!asset.getIdentifier().isBlank()) {
copiedAssetIds.addAll(context.copyAssetFromSource(asset.getIdentifier()).getAssetIdentifiers());
}
}
if (!copiedAssetIds.isEmpty()) {
context.setValueReference(targetStepName, targetPropertyName, new
AssetReference(copiedAssetIds));
}
```
##########
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:
```suggestion
throw new FlowUpdateException("Connector " + connectorId + "
does not support migration.");
```
Nothing wrong with the source flow. Connector just doesn't support migration
##########
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.");
+ }
+
+ try (final NarCloseable ignored =
NarCloseable.withComponentNarLoader(flowController.getExtensionManager(),
rawConnector.getClass(), connectorId)) {
+ if (!migratableConnector.isMigrationSupported(migrationContext)) {
+ throw new FlowUpdateException("Connector " + connectorId + "
does not support migration from the provided source flow.");
+ }
+ }
+
+ // Phase 1: drive migrateConfiguration(...) ->
applyMigratedConfiguration(...). The connector records the
+ // configuration it wants on the other side of migration by mutating
the working configuration clone seeded
+ // into the migration context; the framework then drives
Connector.applyUpdate(workingContext, activeContext)
+ // so the managed Process Group is rebuilt from that merged
configuration. The active configuration itself is
+ // intentionally not written here: the merged configuration is held
until the state phase below succeeds, then
+ // committed by commitMigratedConfiguration(...). That ordering keeps
rollback simple - any failure in either
+ // phase can be recovered by restoring the initial flow because the
active configuration was never mutated.
+ try (final NarCloseable ignored =
NarCloseable.withComponentNarLoader(flowController.getExtensionManager(),
rawConnector.getClass(), connectorId)) {
+
flowController.getConnectorRepository().syncAssetsFromProvider(connector);
+ migratableConnector.migrateConfiguration(migrationContext);
+
+ if (cancellation.getAsBoolean()) {
+ throw new FlowUpdateException("Migration of Connector " +
connectorId + " was cancelled after migrateConfiguration() completed.");
+ }
+ } catch (final FlowUpdateException e) {
+ failMigration(connector, migrationContext);
+ throw e;
+ } catch (final Exception e) {
+ failMigration(connector, migrationContext);
+ throw new FlowUpdateException("Failed to migrate Connector " +
connectorId + " from the provided source flow", e);
+ }
+
+ final ConnectorConfiguration mergedConfiguration;
+ try {
+ mergedConfiguration =
connector.applyMigratedConfiguration(migrationContext.getMergedConfiguration());
+ } catch (final FlowUpdateException e) {
+ failMigration(connector, migrationContext);
+ throw e;
+ } catch (final Exception e) {
+ failMigration(connector, migrationContext);
+ throw new FlowUpdateException("Failed to apply migrated
configuration for Connector " + connectorId, e);
+ }
+
+ // Phase 2: drive migrateState(...) -> drain staged component-state
writes -> apply each entry to the live
+ // StateManager. The managed Process Group has been rebuilt by
applyMigratedConfiguration(...) above, so the
+ // managed components exist before the connector tries to record state
for them. Wrong-phase calls
+ // (setProperties/replaceProperties) now throw IllegalStateException
from the context; updateFlow(...) on
+ // the active flow context still throws via the MigrationFlowContext
wrapper.
+
migrationContext.setPhase(StandardFrameworkConnectorMigrationContext.Phase.STATE);
+ try (final NarCloseable ignored =
NarCloseable.withComponentNarLoader(flowController.getExtensionManager(),
rawConnector.getClass(), connectorId)) {
+ migratableConnector.migrateState(migrationContext);
+
+ if (cancellation.getAsBoolean()) {
+ throw new FlowUpdateException("Migration of Connector " +
connectorId + " was cancelled after migrateState() completed.");
+ }
+ } catch (final FlowUpdateException e) {
+ failMigration(connector, migrationContext);
+ throw e;
+ } catch (final Exception e) {
+ failMigration(connector, migrationContext);
+ throw new FlowUpdateException("Failed to migrate state for
Connector " + connectorId, e);
+ }
+
+ final Map<String, VersionedComponentState> stagedStates =
migrationContext.drainStagedComponentStates();
+ try {
+ applyMigratedComponentStates(connector, stagedStates);
+ } catch (final Exception e) {
+ failMigration(connector, migrationContext);
+ throw new FlowUpdateException("Failed to apply migrated component
state for Connector " + connectorId, e);
+ }
+
+ // Commit: both phases have succeeded, so persist the merged
configuration onto the active configuration.
+ // From this point on the migration is durable across restarts:
inheritConfiguration(...) on the next load will
+ // rebuild the managed Process Group from the committed configuration
via the same Connector.applyUpdate(...)
+ // path used here.
+ try {
+ connector.commitMigratedConfiguration(mergedConfiguration);
+ } catch (final Exception e) {
+ failMigration(connector, migrationContext);
+ throw new FlowUpdateException("Failed to commit migrated
configuration for Connector " + connectorId, e);
+ }
+
+
migrationContext.setPhase(StandardFrameworkConnectorMigrationContext.Phase.COMPLETED);
+
+ // The migration is durable now that the merged configuration is
committed, so notify the
+ // ConnectorConfigurationProvider that the migration completed. The
source Process Group identifier is null for
+ // an uploaded-payload migration. A failure here is logged rather than
rolled back, because rolling back a
+ // committed migration would silently resurrect it on the next restart.
+ try {
+
flowController.getConnectorRepository().notifyMigrationComplete(connectorId,
processGroupId);
+ } catch (final Exception e) {
+ logger.warn("Connector {} migration completed and is durable, but
notifying the configuration provider of completion failed", connectorId, e);
+ }
+
+ // Finalize: the migration is already durable at this point because
the merged configuration has been committed
+ // onto the active configuration. The remaining work is cosmetic -
disabling and renaming the source Process
+ // Group so the operator can tell it apart from the new managed flow.
Failures (or a late cancellation) below
+ // are logged but no longer trigger a rollback because rolling back a
committed migration would leave the
+ // active configuration migrated while the managed Process Group was
reset to the initial flow, which would
+ // silently resurrect the migration on the next restart.
+ if (cancellation.getAsBoolean()) {
+ logger.warn("Migration of Connector {} was cancelled after the
merged configuration was committed; the migration is durable but the source
Process Group was not finalized", connectorId);
+ return;
+ }
+
+ if (sourceProcessGroup != null) {
+ try {
+ disableAndRenameSourceProcessGroup(sourceProcessGroup);
+ } catch (final Exception e) {
+ logger.warn("Failed to finalize source Process Group {} after
a successful migration of Connector {}; manual cleanup may be required",
+ sourceProcessGroup.getIdentifier(), connectorId, e);
+ }
+ }
+ }
+
+ /**
+ * Applies the staged component-state writes drained from {@code
migrateState(...)} onto the live components in
+ * the connector's managed Process Group. For each entry the framework:
+ * <ol>
+ * <li>resolves the managed component by its versioned identifier;</li>
+ * <li>looks up the component's runtime instance identifier and
fetches its {@link StateManager} via
+ * {@link
org.apache.nifi.components.state.StateManagerProvider#getStateManager(String)
+ * StateManagerProvider.getStateManager(...)};</li>
+ * <li>writes the cluster-scope state, when the component declares
{@link Scope#CLUSTER} on its
+ * {@code @Stateful} annotation;</li>
+ * <li>writes the local-scope state for this node, when the component
declares {@link Scope#LOCAL}. The
+ * local-state entry is picked by node ordinal so a clustered
migration installs the right per-node state
+ * on each node, matching the framework's normal versioned-flow
synchronizer.</li>
+ * </ol>
+ * Skips components whose {@code @Stateful} annotation does not declare
the requested scope and logs a warning,
+ * matching the behavior of {@code
StandardVersionedComponentSynchronizer.restoreComponentState(...)}.
+ *
+ * <p>
+ * This loop applies entries sequentially and does not roll back partial
writes on its own. The rollback path that
+ * runs when any entry throws (see {@link #clearConnectorComponentState})
wipes every managed component's
+ * StateManager state, so a partially-applied set of writes is cleared
along with the rest of the migration's
+ * side effects rather than left behind.
+ * </p>
+ */
+ private void applyMigratedComponentStates(final ConnectorNode connector,
final Map<String, VersionedComponentState> stagedStates) throws IOException {
+ if (stagedStates.isEmpty()) {
+ return;
+ }
+
+ final FrameworkFlowContext activeFlowContext =
connector.getActiveFlowContext();
+ if (activeFlowContext == null) {
+ throw new IllegalStateException("Cannot apply migrated component
state for Connector " + connector.getIdentifier()
+ + " because it has no active flow context.");
+ }
+ final ProcessGroup managedGroup =
activeFlowContext.getManagedProcessGroup();
+ if (managedGroup == null) {
+ throw new IllegalStateException("Cannot apply migrated component
state for Connector " + connector.getIdentifier()
+ + " because its managed Process Group does not exist.");
+ }
+
+ final Map<String, ProcessorNode> processorsByVersionedId = new
HashMap<>();
+ for (final ProcessorNode processor : managedGroup.findAllProcessors())
{
+ processor.getVersionedComponentId().ifPresent(versionedId ->
processorsByVersionedId.put(versionedId, processor));
+ }
+
+ final Map<String, ControllerServiceNode> servicesByVersionedId = new
HashMap<>();
+ for (final ControllerServiceNode service :
managedGroup.findAllControllerServices()) {
+ service.getVersionedComponentId().ifPresent(versionedId ->
servicesByVersionedId.put(versionedId, service));
+ }
+
+ final int localNodeOrdinal = flowController.getLocalNodeOrdinal();
+
+ for (final Map.Entry<String, VersionedComponentState> entry :
stagedStates.entrySet()) {
+ final String versionedId = entry.getKey();
+ final VersionedComponentState desiredState = entry.getValue();
+
+ final ProcessorNode processor =
processorsByVersionedId.get(versionedId);
+ if (processor != null) {
+ applyComponentState(processor.getIdentifier(),
processor.getComponent(), desiredState, localNodeOrdinal);
+ continue;
+ }
+ final ControllerServiceNode service =
servicesByVersionedId.get(versionedId);
+ if (service != null) {
+ applyComponentState(service.getIdentifier(),
service.getControllerServiceImplementation(), desiredState, localNodeOrdinal);
+ continue;
+ }
+ logger.warn("Connector {} migrateState() requested state for
component [{}] but no managed Processor or Controller Service with that
versioned identifier exists; skipping",
+ connector.getIdentifier(), versionedId);
+ }
+ }
+
+ private void failMigration(final ConnectorNode connector, final
StandardFrameworkConnectorMigrationContext migrationContext) {
+
migrationContext.setPhase(StandardFrameworkConnectorMigrationContext.Phase.COMPLETED);
+ rollbackMigration(connector, migrationContext);
+ }
+
+ private void applyComponentState(final String runtimeComponentId, final
ConfigurableComponent component, final VersionedComponentState desiredState,
final int localNodeOrdinal)
+ throws IOException {
+ final StateManager stateManager =
flowController.getStateManagerProvider().getStateManager(runtimeComponentId);
+ if (stateManager == null) {
+ logger.warn("No StateManager registered for migrated component
[{}]; skipping state write", runtimeComponentId);
+ return;
+ }
+
+ final Set<Scope> declaredScopes = declaredStatefulScopes(component);
+
+ final Map<String, String> clusterState =
desiredState.getClusterState();
+ if (clusterState != null && !clusterState.isEmpty()) {
+ if (declaredScopes.contains(Scope.CLUSTER)) {
+ stateManager.setState(clusterState, Scope.CLUSTER);
+ } else {
+ logger.warn("Component [{}] does not declare @Stateful scope
CLUSTER; skipping cluster state migration", runtimeComponentId);
+ }
+ }
+
+ final List<VersionedNodeState> localNodeStates =
desiredState.getLocalNodeStates();
+ if (localNodeStates != null && !localNodeStates.isEmpty()) {
+ if (!declaredScopes.contains(Scope.LOCAL)) {
+ logger.warn("Component [{}] does not declare @Stateful scope
LOCAL; skipping local state migration", runtimeComponentId);
+ return;
+ }
+ if (localNodeOrdinal < 0) {
+ logger.warn("Local node ordinal is unknown; skipping local
state migration for component [{}]", runtimeComponentId);
+ return;
+ }
+ if (localNodeOrdinal >= localNodeStates.size()) {
+ logger.warn("Component [{}] migration has {} local node state
entries but this node ordinal is {}; skipping local state migration",
+ runtimeComponentId, localNodeStates.size(),
localNodeOrdinal);
+ return;
+ }
+ final VersionedNodeState nodeState =
localNodeStates.get(localNodeOrdinal);
+ if (nodeState == null || nodeState.getState() == null ||
nodeState.getState().isEmpty()) {
+ return;
+ }
+ stateManager.setState(nodeState.getState(), Scope.LOCAL);
+ }
+ }
+
+ private static Set<Scope> declaredStatefulScopes(final
ConfigurableComponent component) {
+ if (component == null) {
+ return Set.of();
+ }
+ final Stateful stateful =
component.getClass().getAnnotation(Stateful.class);
+ if (stateful == null) {
+ return Set.of();
+ }
+ return EnumSet.copyOf(Arrays.asList(stateful.scopes()));
+ }
+
+ private void rollbackMigration(final ConnectorNode connector, final
StandardFrameworkConnectorMigrationContext migrationContext) {
+ clearConnectorComponentState(connector);
+
+ final Set<String> copiedAssetIds =
migrationContext.getCopiedAssetIds();
+ if (!copiedAssetIds.isEmpty()) {
+ try {
+
flowController.getConnectorRepository().deleteAssets(connector.getIdentifier(),
copiedAssetIds);
+ } catch (final Exception e) {
+ logger.warn("Failed to delete copied assets {} for Connector
{} during migration rollback", copiedAssetIds, connector.getIdentifier(), e);
+ }
+ }
+
+ try {
+ connector.loadInitialFlow();
+ } catch (final FlowUpdateException e) {
+ // Do not propagate the rollback failure: the caller's primary
need is to see why the migration originally
+ // failed, not how the recovery attempt failed. The operator can
read the rollback failure from the log.
+ logger.error("Failed to restore Connector {} to its initial state
after migration failure; manual cleanup may be required",
+ connector.getIdentifier(), e);
+ return;
+ }
+
+
flowController.getConnectorRepository().discardWorkingConfiguration(connector);
+ }
+
+ private void clearConnectorComponentState(final ConnectorNode connector) {
+ // Migration is only permitted on a Connector whose active flow has
not been modified from its initial flow.
+ // Rollback restores that initial flow immediately after this call, so
clearing state for every Processor and
+ // Controller Service currently in the managed flow is acceptable: any
component carried over from the initial
+ // flow is rebuilt fresh by loadInitialFlow(), and any component
introduced by the failed migration is
+ // discarded. Clearing the whole managed flow this way avoids the
brittle approach of trying to map source-flow
+ // versioned identifiers to runtime component identifiers.
+ final FrameworkFlowContext activeFlowContext =
connector.getActiveFlowContext();
+ if (activeFlowContext == null) {
+ return;
+ }
+
+ final ProcessGroup managedGroup =
activeFlowContext.getManagedProcessGroup();
+ if (managedGroup == null) {
+ return;
+ }
+
+ for (final ProcessorNode processor : managedGroup.findAllProcessors())
{
+ clearComponentState(processor.getIdentifier());
+ }
+ for (final ControllerServiceNode controllerService :
managedGroup.findAllControllerServices()) {
+ clearComponentState(controllerService.getIdentifier());
+ }
+ }
+
+ private void clearComponentState(final String componentIdentifier) {
+ final StateManager stateManager =
flowController.getStateManagerProvider().getStateManager(componentIdentifier);
+ if (stateManager == null) {
+ return;
+ }
+
+ try {
+ stateManager.clear(Scope.LOCAL);
+ } catch (final Exception e) {
+ logger.warn("Failed to clear LOCAL state for component {} during
migration rollback", componentIdentifier, e);
+ }
+
+ try {
+ stateManager.clear(Scope.CLUSTER);
+ } catch (final Exception e) {
+ logger.warn("Failed to clear CLUSTER state for component {} during
migration rollback", componentIdentifier, e);
+ }
+ }
+
+ private ConnectorNode getRequiredConnector(final String connectorId) {
+ final ConnectorNode connector =
flowController.getConnectorRepository().getConnector(connectorId,
ConnectorSyncMode.LOCAL_ONLY);
+ if (connector == null) {
+ throw new IllegalArgumentException("Could not find Connector with
ID " + connectorId);
+ }
+
+ return connector;
+ }
+
+ private ProcessGroup getRequiredSourceProcessGroup(final String
processGroupId) {
+ final ProcessGroup rootGroup =
flowController.getFlowManager().getRootGroup();
+ final ProcessGroup processGroup = rootGroup == null ? null :
rootGroup.findProcessGroup(processGroupId);
+ if (processGroup == null) {
+ throw new IllegalArgumentException("Could not find Process Group
with ID " + processGroupId);
+ }
+
+ return processGroup;
+ }
+
+ private List<ProcessGroup> getCandidateSourceGroups() {
+ final FlowManager flowManager = flowController.getFlowManager();
+ final ProcessGroup rootGroup = flowManager.getRootGroup();
+ if (rootGroup == null) {
+ return Collections.emptyList();
+ }
+
+ return new ArrayList<>(rootGroup.findAllProcessGroups());
+ }
+
+ private String getIneligibilityReason(final ConnectorNode connector, final
ProcessGroup processGroup) {
+ final VersionControlInformation versionControlInformation =
processGroup.getVersionControlInformation();
+ if (versionControlInformation == null) {
+ return "The Process Group is not under version control.";
+ }
+
+ if (processGroup.getConnectorIdentifier().isPresent()) {
+ return "The Process Group is managed by another Connector and
cannot be used as a migration source.";
+ }
+
+ final VersionedExternalFlow sourceFlow =
obtainSourceFlowForListing(processGroup);
+
+ final List<String> stateIneligibilityReasons =
getStateIneligibilityReasons(processGroup, versionControlInformation,
sourceFlow);
+ if (!stateIneligibilityReasons.isEmpty()) {
+ // Each entry describes a distinct condition the user can address.
Joining with newlines keeps the
+ // individual reasons identifiable (e.g. "There are 8 running
Processors.\nThere are 3 queued FlowFiles.")
+ // rather than collapsing them into a single run-on sentence when
they are surfaced to the operator.
+ return String.join(System.lineSeparator(),
stateIneligibilityReasons);
+ }
+
+ if (!isConnectorMigrationSupported(connector, sourceFlow)) {
+ return "The Connector does not support migration from this Process
Group.";
+ }
+
+ return null;
+ }
+
+ /**
+ * Reports every condition that currently prevents the Process Group from
being used as a migration source.
+ * Reasons describe conditions the user can remediate (stopping
processors, draining queues, refreshing version
+ * control, etc.) without changing the structural compatibility of the
source flow. The returned list contains
+ * one entry per applicable condition so the user can address them
together; an empty list means the Process
+ * Group is in a state that allows migration. Callers must ensure the
supplied {@code versionControlInformation}
+ * matches the {@code processGroup}; this method does not tolerate a null
value.
+ *
+ * @return ordered list of state-based ineligibility reasons, or an empty
list if the Process Group is in a state
+ * that allows migration
+ */
+ private List<String> getStateIneligibilityReasons(final ProcessGroup
processGroup, final VersionControlInformation versionControlInformation, final
VersionedExternalFlow sourceFlow) {
+ final List<String> reasons = new ArrayList<>();
+
+ final VersionedFlowStatus versionedFlowStatus =
versionControlInformation.getStatus();
+ final VersionedFlowState state = versionedFlowStatus == null ? null :
versionedFlowStatus.getState();
+ if (state != VersionedFlowState.UP_TO_DATE) {
+ reasons.add(describeIneligibleVersionedFlowState(state));
+ }
+
+ final int runningProcessorCount = countRunningProcessors(processGroup);
+ if (runningProcessorCount > 0) {
+ reasons.add("There " + pluralizeIsAre(runningProcessorCount) + " "
+ runningProcessorCount + " running or enabled "
+ + pluralizeNoun(runningProcessorCount, "processor",
"processors") + ".");
+ }
+
+ final int enabledServiceCount =
countEnabledControllerServices(processGroup);
+ if (enabledServiceCount > 0) {
+ reasons.add("There " + pluralizeIsAre(enabledServiceCount) + " " +
enabledServiceCount + " enabled controller "
+ + pluralizeNoun(enabledServiceCount, "service",
"services") + ".");
+ }
+
+ final long queuedFlowFileCount = countQueuedFlowFiles(processGroup);
+ if (queuedFlowFileCount > 0) {
+ reasons.add("There " + pluralizeIsAre(queuedFlowFileCount) + " " +
queuedFlowFileCount + " queued "
+ + pluralizeNoun(queuedFlowFileCount, "FlowFile",
"FlowFiles") + ".");
+ }
+
+ final int externalControllerServiceCount =
countExternalControllerServices(sourceFlow);
+ if (externalControllerServiceCount > 0) {
+ reasons.add("The Process Group references " +
externalControllerServiceCount + " controller "
+ + pluralizeNoun(externalControllerServiceCount, "service",
"services") + " from outside the Process Group.");
+ }
+
+ return reasons;
+ }
+
+ private static String pluralizeIsAre(final long count) {
Review Comment:
While correct grammatically does this make log analysis harder?
--
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]
