markap14 commented on code in PR #11240:
URL: https://github.com/apache/nifi/pull/11240#discussion_r3500633980
##########
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:
On (2): good point — fixed. The uploaded-payload path now runs the same
synchronous readiness check (connector state + `matchesInitialFlow`) as the
local-source path, via `verifyConnectorReadyForMigration`, so the user sees the
failure on submission instead of several poll cycles later.
On (1): there's no per-Connector migration lock today. Submissions are
serialized by the revision-based `withWriteLock` and re-verified both
synchronously at submission and again at the start of the async task, so two
requests that share a starting revision can't both proceed. The residual window
is two requests that each pick up a fresh revision and both pass verification
before either commits `applyMigratedConfiguration`. This matches how ordinary
async connector updates behave, so I've left it as-is for now rather than
introducing a manager-level lock or an explicit `MIGRATING` `ConnectorState`.
--
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]
