rfellows commented on code in PR #11240: URL: https://github.com/apache/nifi/pull/11240#discussion_r3469197192
########## nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/MigrationRequestEndpointMerger.java: ########## @@ -0,0 +1,107 @@ +/* + * 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.cluster.coordination.http.endpoints; + +import org.apache.nifi.cluster.manager.NodeResponse; +import org.apache.nifi.cluster.protocol.NodeIdentifier; +import org.apache.nifi.web.api.dto.MigrationRequestDTO; +import org.apache.nifi.web.api.dto.MigrationUpdateStepDTO; +import org.apache.nifi.web.api.entity.MigrationRequestEntity; + +import java.net.URI; +import java.util.Date; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.regex.Pattern; + +public class MigrationRequestEndpointMerger extends AbstractSingleEntityEndpoint<MigrationRequestEntity> { + public static final Pattern MIGRATION_REQUEST_URI_PATTERN = + Pattern.compile("/nifi-api/connectors/[a-f0-9\\-]{36}/migration-requests(/[a-f0-9\\-]{36})?"); + + @Override + protected Class<MigrationRequestEntity> getEntityClass() { + return MigrationRequestEntity.class; + } + + @Override + public boolean canHandle(final URI uri, final String method) { + return MIGRATION_REQUEST_URI_PATTERN.matcher(uri.getPath()).matches(); + } + + @Override + protected void mergeResponses(final MigrationRequestEntity clientEntity, final Map<NodeIdentifier, MigrationRequestEntity> entityMap, + final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses) { + final MigrationRequestDTO clientRequest = clientEntity.getRequest(); + if (clientRequest == null) { + return; + } + + final List<MigrationUpdateStepDTO> clientUpdateSteps = clientRequest.getUpdateSteps() == null ? List.of() : clientRequest.getUpdateSteps(); + + for (final Map.Entry<NodeIdentifier, MigrationRequestEntity> nodeEntry : entityMap.entrySet()) { + final NodeIdentifier nodeId = nodeEntry.getKey(); + final MigrationRequestDTO nodeRequest = nodeEntry.getValue().getRequest(); + if (nodeRequest == null) { + continue; + } + + clientRequest.setComplete(clientRequest.isComplete() && nodeRequest.isComplete()); + + if (nodeRequest.getFailureReason() != null) { + final String nodeReason = "Node " + nodeId.getApiAddress() + ":" + nodeId.getApiPort() + ": " + nodeRequest.getFailureReason(); + final String existingReason = clientRequest.getFailureReason(); + clientRequest.setFailureReason(existingReason == null ? nodeReason : existingReason + "; " + nodeReason); + } + + final Date clientLastUpdated = clientRequest.getLastUpdated(); + final Date nodeLastUpdated = nodeRequest.getLastUpdated(); + if (nodeLastUpdated != null && (clientLastUpdated == null || nodeLastUpdated.before(clientLastUpdated))) { Review Comment: The condition uses `nodeLastUpdated.before(clientLastUpdated)`, which means the merged `lastUpdated` will reflect the **earliest** timestamp across nodes. The neighboring merges (`setComplete = AND`, `setPercentCompleted = min`) are "worst-of" for completion progress, but for an activity timestamp a polling client would typically expect the **most recent** update across the cluster — otherwise the merged value can appear to go backward as more nodes report in. Is the intent here to surface the oldest timestamp (and if so, what does that represent for the client), or should this be `nodeLastUpdated.after(clientLastUpdated)`? `MigrationRequestEndpointMergerTest` doesn't currently assert anything about the merged `lastUpdated`, so either direction would pass today — happy to suggest a test once the intent is confirmed. ########## nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java: ########## @@ -664,7 +675,16 @@ private void waitForState(final ConnectorNode connector, final Set<ConnectorStat int iterations = 0; final long startNanos = System.nanoTime(); while (true) { - final ConnectorState clusterState = requestReplicator.getState(connector.getIdentifier()); + final ConnectorState clusterState; + try { + clusterState = requestReplicator.getState(connector.getIdentifier()); + } catch (final IOException e) { + final long elapsedSeconds = Duration.ofNanos(System.nanoTime() - startNanos).toSeconds(); + logger.warn("Failed to retrieve cluster state for {} while waiting for update completion; elapsed time = {} secs", connector, elapsedSeconds, e); + Thread.sleep(Duration.ofSeconds(1)); + continue; Review Comment: The new `catch (IOException)` block logs and `continue`s, but the only place `iterations` is incremented (and therefore the only termination opportunity short of a non-`IOException` state outcome) is the `allowableStates` branch below — which isn't reached if `requestReplicator.getState(...)` keeps throwing. If the request replicator is persistently unhealthy, would this loop ever exit, or would the caller block indefinitely? Was an upper bound intended here — e.g. using `startNanos` to fail after a max duration, or bailing out after N consecutive `IOException`s? Happy to be wrong if there's an outer cancellation/timeout I'm not seeing. ########## nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ConnectorResource.java: ########## @@ -2353,6 +2403,342 @@ public Response getConnectorRemoteProcessGroupStatusHistory( return generateOkResponse(entity).build(); } + @GET + @Consumes(MediaType.WILDCARD) + @Produces(MediaType.APPLICATION_JSON) + @Path("{id}/migration-sources") + @Operation( + summary = "Lists the Versioned Process Groups that the Connector can be migrated from", + responses = { + @ApiResponse(responseCode = "200", content = @Content(schema = @Schema(implementation = VersionedFlowMigrationSourcesEntity.class))), + @ApiResponse(responseCode = "400", description = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(responseCode = "401", description = "Client could not be authenticated."), + @ApiResponse(responseCode = "403", description = "Client is not authorized to make this request."), + @ApiResponse(responseCode = "404", description = "The specified resource could not be found."), + @ApiResponse(responseCode = "409", description = "The request was valid but NiFi was not in the appropriate state to process it.") + }, + security = { + @SecurityRequirement(name = "Read - /connectors/{uuid}") + } + ) + public Response getMigrationSources(@PathParam("id") final String connectorId) { + authorizeReadConnector(connectorId); + + if (isReplicateRequest()) { + return replicate(HttpMethod.GET); + } + + final VersionedFlowMigrationSourcesEntity entity = serviceFacade.getConnectorMigrationSources(connectorId); + return generateOkResponse(entity).build(); + } + + @POST + @Consumes(MediaType.APPLICATION_OCTET_STREAM) + @Produces(MediaType.APPLICATION_JSON) + @Path("{id}/migration-payloads") + @Operation( + summary = "Uploads a flow snapshot payload for a later Connector migration request", + responses = { + @ApiResponse(responseCode = "200", content = @Content(schema = @Schema(implementation = MigrationPayloadEntity.class))), + @ApiResponse(responseCode = "400", description = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(responseCode = "401", description = "Client could not be authenticated."), + @ApiResponse(responseCode = "403", description = "Client is not authorized to make this request."), + @ApiResponse(responseCode = "404", description = "The specified resource could not be found."), + @ApiResponse(responseCode = "409", description = "The request was valid but NiFi was not in the appropriate state to process it.") + }, + security = { + @SecurityRequirement(name = "Write - /connectors/{uuid}") + } + ) + public Response createMigrationPayload( + @PathParam("id") final String connectorId, + @Parameter(description = "The migration payload snapshot", required = true) final InputStream payloadContents) throws IOException { + if (payloadContents == null) { + throw new IllegalArgumentException("Migration payload contents must be specified."); + } + + final NiFiUser currentUser = NiFiUserUtils.getNiFiUser(); + serviceFacade.authorizeAccess(lookup -> { + final Authorizable connector = lookup.getConnector(connectorId); + connector.authorize(authorizer, RequestAction.WRITE, currentUser); + }); + + if (isReplicateRequest()) { + final String uploadRequestId = UUID.randomUUID().toString(); + final UploadRequest<MigrationPayloadEntity> uploadRequest = new UploadRequest.Builder<MigrationPayloadEntity>() + .user(currentUser) + .filename("migration-payload.json") + .identifier(uploadRequestId) + .contents(payloadContents) + .forwardRequestHeaders(getHeaders()) + .header(CONTENT_TYPE_HEADER, UPLOAD_CONTENT_TYPE) + .header(RequestReplicationHeader.CLUSTER_ID_GENERATION_SEED.getHeader(), uploadRequestId) + .exampleRequestUri(getAbsolutePath()) + .responseClass(MigrationPayloadEntity.class) + .successfulResponseStatus(HttpResponseStatus.OK.getCode()) + .build(); + final MigrationPayloadEntity entity = uploadRequestReplicator.upload(uploadRequest); + return generateOkResponse(entity).build(); + } + + final RegisteredFlowSnapshot flowSnapshot; + try { + flowSnapshot = OBJECT_MAPPER.readValue(payloadContents, RegisteredFlowSnapshot.class); + } catch (final IOException e) { + throw new IllegalArgumentException("Deserialization of uploaded migration payload failed", e); + } + + // Bundle discovery is deferred to the asynchronous migration task in performAsyncMigration so that + // the upload thread does not block on extension-manager work. + final String payloadId = getIdGenerationSeed().orElseGet(this::generateUuid); + migrationPayloadsById.put(payloadId, new MigrationPayloadEntry(flowSnapshot, System.currentTimeMillis())); + final MigrationPayloadDTO migrationPayload = new MigrationPayloadDTO(); + migrationPayload.setPayloadId(payloadId); + + final MigrationPayloadEntity entity = new MigrationPayloadEntity(); + entity.setPayload(migrationPayload); + return generateOkResponse(entity).build(); + } + + @POST + @Consumes(MediaType.APPLICATION_JSON) + @Produces(MediaType.APPLICATION_JSON) + @Path("/{id}/migration-requests") + @Operation( + summary = "Creates a Connector migration request", + responses = { + @ApiResponse(responseCode = "200", content = @Content(schema = @Schema(implementation = MigrationRequestEntity.class))), + @ApiResponse(responseCode = "400", description = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(responseCode = "401", description = "Client could not be authenticated."), + @ApiResponse(responseCode = "403", description = "Client is not authorized to make this request."), + @ApiResponse(responseCode = "404", description = "The specified resource could not be found."), + @ApiResponse(responseCode = "409", description = "The request was valid but NiFi was not in the appropriate state to process it.") + }, + security = { + @SecurityRequirement(name = "Write - /connectors/{uuid}") + } + ) + public Response createMigrationRequest(@PathParam("id") final String connectorId, final MigrationRequestEntity requestEntity) { + if (requestEntity == null || requestEntity.getRequest() == null) { + throw new IllegalArgumentException("Migration request must be specified."); + } + + final MigrationRequestDTO request = requestEntity.getRequest(); + if (!connectorId.equals(request.getConnectorId())) { + throw new IllegalArgumentException("Connector identifier in the request must match the identifier provided in the URL."); + } + + final boolean hasLocalSource = request.getLocalSource() != null && StringUtils.isNotBlank(request.getLocalSource().getProcessGroupId()); + final boolean hasPayload = StringUtils.isNotBlank(request.getPayloadId()); + if (hasLocalSource == hasPayload) { + throw new IllegalArgumentException("Migration request must specify exactly one source: either a local Process Group or an uploaded payload identifier."); + } + + if (hasPayload && !migrationPayloadsById.containsKey(request.getPayloadId())) { + throw new ResourceNotFoundException("No uploaded migration payload exists with identifier " + request.getPayloadId()); + } + + // Reject the request when not all cluster nodes are connected, mirroring the asset-upload guard, + // because component state and assets cannot be synchronized to disconnected nodes after migration. + final ClusterCoordinator clusterCoordinator = getClusterCoordinator(); + if (clusterCoordinator != null) { + final Set<NodeIdentifier> disconnectedNodes = clusterCoordinator.getNodeIdentifiers(NodeConnectionState.CONNECTING, NodeConnectionState.DISCONNECTED, NodeConnectionState.DISCONNECTING); + if (!disconnectedNodes.isEmpty()) { + throw new IllegalStateException("Cannot start a Connector migration because the following %s nodes are not currently connected: %s" + .formatted(disconnectedNodes.size(), disconnectedNodes)); + } + } + + if (isReplicateRequest()) { + return replicate(HttpMethod.POST, requestEntity); + } + + final NiFiUser user = NiFiUserUtils.getNiFiUser(); + return withWriteLock( + serviceFacade, + requestEntity, + lookup -> { + final Authorizable connector = lookup.getConnector(connectorId); + connector.authorize(authorizer, RequestAction.WRITE, user); + }, + () -> { + if (hasLocalSource) { + serviceFacade.verifyCanMigrateConnector(connectorId, request.getLocalSource().getProcessGroupId()); Review Comment: Two related questions about concurrency on this submission path: 1. `withWriteLock` is revision-based and the verifier runs synchronously, but the actual migration is dispatched to `migrationRequestManager` and runs asynchronously. Is there a per-Connector serialization mechanism I'm missing that prevents two near-simultaneous submissions against the same target Connector from both passing `verifyTargetIsAtInitialFlow` and then racing in `applyMigratedConfiguration`? If not, would a per-Connector lock at the manager level — or introducing an explicit `MIGRATING` `ConnectorState` to gate transitions — be worth considering? The asymmetric-failure cluster IT exercises per-node divergence but doesn't appear to exercise two parallel requests against the same Connector. 2. The `verifyCanMigrateConnector` call is intentionally guarded by `hasLocalSource`, so for uploaded-payload sources the structural / `matchesInitialFlow` checks only fire inside the async task and surface several poll cycles later. Was that intentional? Would it be reasonable to run an equivalent synchronous verify (Connector state + `matchesInitialFlow`) for the uploaded-payload case too, so the user sees the failure on submission rather than after polling? ########## nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java: ########## @@ -6290,7 +6351,7 @@ public RegisteredFlowSnapshot getCurrentFlowSnapshotByGroupId(final String proce .mapInstanceIdentifiers(false) .mapControllerServiceReferencesToVersionedId(true) .mapFlowRegistryClientId(false) - .mapAssetReferences(false) + .mapAssetReferences(true) Review Comment: Flipping `mapAssetReferences` from `false` → `true` on the shared `getCurrentFlowSnapshotByGroupId` mapping options changes the result for **every** caller of this path, not just the migration source-flow capture — e.g. process-group export and any other consumer of `buildFlowSnapshot` via this method. Is the intent to start including asset references in those snapshots universally, or is the change really only meant for migration? If the latter, would it make sense to route the migration-source capture through a dedicated method (or add an opt-in parameter), so the existing export consumers don't inadvertently start emitting NiFi-internal asset identifiers that may not resolve on a different instance? ########## nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java: ########## @@ -287,9 +290,19 @@ public ConnectorSyncResult syncConnector(final VersionedConnector versionedConne connector.restoreTroubleshootingState(); } + // Asset cleanup is intentionally skipped while the Connector is in Troubleshooting mode. Troubleshooting is + // a transient state in which the user may have uploaded Assets that the experimental flow does not yet + // reference and may have temporarily removed references to the authoritative Assets. Deleting unreferenced + // Assets here would discard Assets the user still needs. Orphaned Assets are reclaimed by the next + // authoritative configuration change, which is only permitted once Troubleshooting mode has ended. return ConnectorSyncResult.syncedConfigUnchanged(connector, effectiveScheduledState); } + // Clean up assets after configuration has been inherited so that the Connector's flow contexts reflect the + // synchronized configuration; performing this before synchronization would treat still-referenced assets as + // unreferenced because the managed flow has not yet been populated. + cleanUpAssets(connector); Review Comment: `cleanUpAssets(connector)` is now invoked on every successful `syncConnector(...)` (a broader trigger than before this PR), and together with the new `collectReferencedParameterContextAssetIds` it's necessary for asset rollback to behave correctly after a failed migration. I want to confirm the lifecycle relationship to migration: is `syncConnector(...)` guaranteed never to run concurrently with an in-flight `applyMigratedConfiguration(...)` on the same Connector? If a sync ever interleaved with the migration's mid-Phase-2 state — after the new working configuration is built but before it's committed — could `cleanUpAssets` see the newly-copied assets as unreferenced and delete them? Even if that interleaving isn't reachable today, a one-line comment locking the invariant in place (or a targeted test) would make this much harder to regress. ########## nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/audit/ConnectorAuditor.java: ########## @@ -372,6 +372,29 @@ public void applyConnectorUpdateAdvice(final ProceedingJoinPoint proceedingJoinP } } + @Around("within(org.apache.nifi.web.dao.ConnectorDAO+) && " + + "execution(void migrateFromVersionedFlow(java.lang.String, java.lang.String, org.apache.nifi.flow.VersionedExternalFlow, java.util.function.BooleanSupplier)) && " + + "args(connectorId, processGroupId, sourceFlow, cancellationCheck) && " + + "target(connectorDAO)") + public void migrateConnectorAdvice(final ProceedingJoinPoint proceedingJoinPoint, final String connectorId, final String processGroupId, + final Object sourceFlow, final Object cancellationCheck, final ConnectorDAO connectorDAO) throws Throwable { + final ConnectorNode connector = connectorDAO.getConnector(connectorId); + + proceedingJoinPoint.proceed(); + + if (isAuditable()) { + final FlowChangeConfigureDetails actionDetails = new FlowChangeConfigureDetails(); + actionDetails.setName("Connector migrated from Versioned Process Group"); + actionDetails.setPreviousValue(null); + actionDetails.setValue(processGroupId == null ? "Uploaded payload" : processGroupId); + + final Action action = generateAuditRecord(connector, Operation.Configure, actionDetails); Review Comment: Migration is recorded as `Operation.Configure`, but semantically it's distinct from an ordinary configuration change — it can copy assets, write component state, install a managed Process Group, disable and rename the source PG, and is effectively irreversible. Operators filtering the audit log for migration events would currently have to look for `Operation.Configure` rows whose details `name` starts with "Connector migrated from…". Would it be reasonable to introduce a dedicated `Operation.Migrate` (alongside the existing operations) and use it here, so migrations surface as a first-class event type in the audit trail? ########## nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorMigrationManager.java: ########## @@ -0,0 +1,758 @@ +/* + * 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.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 migrateFromVersionedFlow(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); + 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); + + // 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 is empty, so every Processor and Controller Review Comment: The comment here states "Migration is only permitted on a Connector whose active flow is empty", but the precondition check `verifyTargetIsAtInitialFlow` (~L603) only asserts `matchesInitialFlow()` — which is true whenever the live managed flow equals `getInitialFlow()`, including the case where the initial flow itself declares processors or controller services. If a `MigratableConnector` implementation ever ships with a non-empty `getInitialFlow()` containing stateful components, would a Phase 2 failure here wipe state belonging to those originally-declared components? Is the empty-initial-flow assumption something we want to: 1. tighten in `verifyTargetIsAtInitialFlow` (also assert the managed flow is structurally empty), 2. track explicitly (snapshot pre-migration component IDs and only clear "new since snapshot"), or 3. document as a contract requirement on `MigratableConnector` / the developer guide? If today's expectation is "(3) by convention", a one-line note in the developer guide and a code comment here would make the contract clear to future implementers. ########## nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ConnectorResource.java: ########## @@ -2353,6 +2403,342 @@ public Response getConnectorRemoteProcessGroupStatusHistory( return generateOkResponse(entity).build(); } + @GET + @Consumes(MediaType.WILDCARD) + @Produces(MediaType.APPLICATION_JSON) + @Path("{id}/migration-sources") + @Operation( + summary = "Lists the Versioned Process Groups that the Connector can be migrated from", + responses = { + @ApiResponse(responseCode = "200", content = @Content(schema = @Schema(implementation = VersionedFlowMigrationSourcesEntity.class))), + @ApiResponse(responseCode = "400", description = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(responseCode = "401", description = "Client could not be authenticated."), + @ApiResponse(responseCode = "403", description = "Client is not authorized to make this request."), + @ApiResponse(responseCode = "404", description = "The specified resource could not be found."), + @ApiResponse(responseCode = "409", description = "The request was valid but NiFi was not in the appropriate state to process it.") + }, + security = { + @SecurityRequirement(name = "Read - /connectors/{uuid}") + } + ) + public Response getMigrationSources(@PathParam("id") final String connectorId) { + authorizeReadConnector(connectorId); + + if (isReplicateRequest()) { + return replicate(HttpMethod.GET); + } + + final VersionedFlowMigrationSourcesEntity entity = serviceFacade.getConnectorMigrationSources(connectorId); + return generateOkResponse(entity).build(); + } + + @POST + @Consumes(MediaType.APPLICATION_OCTET_STREAM) + @Produces(MediaType.APPLICATION_JSON) + @Path("{id}/migration-payloads") + @Operation( + summary = "Uploads a flow snapshot payload for a later Connector migration request", + responses = { + @ApiResponse(responseCode = "200", content = @Content(schema = @Schema(implementation = MigrationPayloadEntity.class))), + @ApiResponse(responseCode = "400", description = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(responseCode = "401", description = "Client could not be authenticated."), + @ApiResponse(responseCode = "403", description = "Client is not authorized to make this request."), + @ApiResponse(responseCode = "404", description = "The specified resource could not be found."), + @ApiResponse(responseCode = "409", description = "The request was valid but NiFi was not in the appropriate state to process it.") + }, + security = { + @SecurityRequirement(name = "Write - /connectors/{uuid}") + } + ) + public Response createMigrationPayload( + @PathParam("id") final String connectorId, + @Parameter(description = "The migration payload snapshot", required = true) final InputStream payloadContents) throws IOException { + if (payloadContents == null) { + throw new IllegalArgumentException("Migration payload contents must be specified."); + } + + final NiFiUser currentUser = NiFiUserUtils.getNiFiUser(); + serviceFacade.authorizeAccess(lookup -> { + final Authorizable connector = lookup.getConnector(connectorId); + connector.authorize(authorizer, RequestAction.WRITE, currentUser); + }); + + if (isReplicateRequest()) { + final String uploadRequestId = UUID.randomUUID().toString(); + final UploadRequest<MigrationPayloadEntity> uploadRequest = new UploadRequest.Builder<MigrationPayloadEntity>() + .user(currentUser) + .filename("migration-payload.json") + .identifier(uploadRequestId) + .contents(payloadContents) + .forwardRequestHeaders(getHeaders()) + .header(CONTENT_TYPE_HEADER, UPLOAD_CONTENT_TYPE) + .header(RequestReplicationHeader.CLUSTER_ID_GENERATION_SEED.getHeader(), uploadRequestId) + .exampleRequestUri(getAbsolutePath()) + .responseClass(MigrationPayloadEntity.class) + .successfulResponseStatus(HttpResponseStatus.OK.getCode()) + .build(); + final MigrationPayloadEntity entity = uploadRequestReplicator.upload(uploadRequest); + return generateOkResponse(entity).build(); + } + + final RegisteredFlowSnapshot flowSnapshot; + try { + flowSnapshot = OBJECT_MAPPER.readValue(payloadContents, RegisteredFlowSnapshot.class); Review Comment: The payload upload reads the JSON in full via `OBJECT_MAPPER.readValue(payloadContents, ...)` and pins the deserialized snapshot in `migrationPayloadsById` for up to ~10 minutes, but there's no `MaxLengthInputStream` wrap or other size cap on the upload. The asset upload path in this same file caps at `MAX_ASSET_SIZE_BYTES = 1 GB` for comparable resource reasons. Is the expectation that the 10-min TTL + per-Connector WRITE authorization is enough bounding, or would a `MaxLengthInputStream` wrap here be worth adding (probably with a tighter cap than the 1 GB asset limit, since exported flows are typically a couple orders of magnitude smaller)? -- 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]
