You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "pvillard31 (via GitHub)" <gi...@apache.org> on 2026/03/11 12:59:26 UTC
[PR] NIFI-15696 - Add Rebase operation for versioned process groups [nifi]
pvillard31 opened a new pull request, #10994:
URL: https://github.com/apache/nifi/pull/10994
# Summary
NIFI-15696 - Add Rebase operation for versioned process groups
When a versioned process group is at version N with local modifications (state LOCALLY_MODIFIED_AND_STALE), upgrading to a newer version requires reverting all local changes first, then changing version, then manually re-applying every customization. This is error-prone, causes unnecessary downtime, and discourages users from upgrading.
This change adds a "Rebase" operation that upgrades a versioned process group to a newer registry version while automatically preserving compatible local changes — analogous to git rebase.
**How it works:**
The rebase operation has two phases:
- Analysis (read-only): Computes a three-way diff between the local flow, the base registry version (N), and the target registry version. Each local change is classified as Compatible (will be preserved), Conflicting (same field changed both locally and upstream to different values), or Unsupported (change type not yet supported by a rebase handler). Rebase is only allowed when all local changes are compatible.
- Execution: Deep-clones the target version snapshot, overlays compatible local changes onto it to produce a merged snapshot, then synchronizes the process group to the merged snapshot using the existing FlowUpdateResource async infrastructure. The VCI is set to the clean target version so that subsequent state checks correctly report LOCALLY_MODIFIED.
Supported change types (for this first iteration, this will evolve over time):
- POSITION_CHANGED, SIZE_CHANGED, BENDPOINTS_CHANGED — cosmetic, always compatible (local wins)
- PROPERTY_CHANGED, PROPERTY_ADDED — compatible unless upstream changed the same property on the same component to a different value (convergent changes where both sides set the same value are accepted)
- COMMENTS_CHANGED — compatible unless upstream also changed comments on the same component
Unsupported change types (examples: COMPONENT_ADDED, COMPONENT_REMOVED) block the rebase with a clear message. The handler architecture is extensible — adding support for a new change type requires only implementing a single RebaseHandler class.
We follow a similar approach as what already exists for the new API endpoints:
- GET /versions/rebase-analysis/process-groups/{id}?targetVersion={version}
returns the three-way analysis
- POST /versions/rebase-requests/process-groups/{id}
initiates async rebase execution
- GET/DELETE /versions/rebase-requests/{id}
polling and cleanup (reuses FlowUpdateResource)
# Tracking
Please complete the following tracking steps prior to pull request creation.
### Issue Tracking
- [ ] [Apache NiFi Jira](https://issues.apache.org/jira/browse/NIFI) issue created
### Pull Request Tracking
- [ ] Pull Request title starts with Apache NiFi Jira issue number, such as `NIFI-00000`
- [ ] Pull Request commit message starts with Apache NiFi Jira issue number, as such `NIFI-00000`
- [ ] Pull request contains [commits signed](https://docs.github.com/en/authentication/managing-commit-signature-verification/signing-commits) with a registered key indicating `Verified` status
### Pull Request Formatting
- [ ] Pull Request based on current revision of the `main` branch
- [ ] Pull Request refers to a feature branch with one commit containing changes
# Verification
Please indicate the verification steps performed prior to pull request creation.
### Build
- [ ] Build completed using `./mvnw clean install -P contrib-check`
- [ ] JDK 21
- [ ] JDK 25
### Licensing
- [ ] New dependencies are compatible with the [Apache License 2.0](https://apache.org/licenses/LICENSE-2.0) according to the [License Policy](https://www.apache.org/legal/resolved.html)
- [ ] New dependencies are documented in applicable `LICENSE` and `NOTICE` files
### Documentation
- [ ] Documentation formatting appears as expected in rendered files
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15696 - Add Rebase operation for versioned process groups [nifi]
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory merged PR #10994:
URL: https://github.com/apache/nifi/pull/10994
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15696 - Add Rebase operation for versioned process groups [nifi]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #10994:
URL: https://github.com/apache/nifi/pull/10994#issuecomment-4930886064
Automated review is marking this PR as stale due to lack of updates in the past four months. This PR will be closed in 15 days if the stale label is not removed. This stale label and automated closure does not indicate a judgement of the PR, just lack of reviewer bandwidth and helps us keep the PR queue more manageable. If you would like this PR re-opened you can do so and a committer can remove the stale label. Or you can open a new PR. Try to help review other PRs to increase PR review bandwidth which in turn helps yours.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15696 - Add Rebase operation for versioned process groups [nifi]
Posted by "pvillard31 (via GitHub)" <gi...@apache.org>.
pvillard31 commented on PR #10994:
URL: https://github.com/apache/nifi/pull/10994#issuecomment-5033694395
Thanks for the review @exceptionfactory - it had been a while since I submitted the PR so I addressed your feedback + the rebase on a commit and I made some additional changes in another commit on top.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15696 - Add Rebase operation for versioned process groups [nifi]
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #10994:
URL: https://github.com/apache/nifi/pull/10994#discussion_r3617439038
##########
nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/VersionsResource.java:
##########
@@ -996,6 +998,186 @@ public Response deleteRevertRequest(
return deleteFlowUpdateRequest("revert-requests", revertRequestId, disconnectedNodeAcknowledged);
}
+ @GET
+ @Consumes(MediaType.WILDCARD)
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("rebase-analysis/process-groups/{id}")
+ @Operation(
+ summary = "Gets a Rebase Analysis for a Process Group",
+ responses = {
+ @ApiResponse(responseCode = "200", content = @Content(schema = @Schema(implementation = RebaseAnalysisEntity.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.")
+ },
+ description = "For a Process Group that is under Version Control, this will perform a rebase analysis by comparing "
+ + "local modifications against upstream changes between the current version and the specified target version. "
+ + "The analysis determines whether a rebase is allowed or if there are conflicts.",
Review Comment:
Multiline strings can be used instead of concatenation
##########
nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseEngine.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * 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.registry.flow.diff;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.flow.VersionedProcessGroup;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class RebaseEngine {
Review Comment:
I recommend defining an interface and making this the standard implementation to clarify the public contract for engine operations
##########
nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseAnalysis.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.registry.flow.diff;
+
+import org.apache.nifi.flow.VersionedProcessGroup;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+
+public class RebaseAnalysis {
+
+ private final List<ClassifiedDifference> classifiedLocalChanges;
+ private final Set<FlowDifference> upstreamDifferences;
+ private final boolean rebaseAllowed;
+ private final String analysisFingerprint;
+ private final VersionedProcessGroup mergedSnapshot;
+
+ public RebaseAnalysis(final List<ClassifiedDifference> classifiedLocalChanges, final Set<FlowDifference> upstreamDifferences,
+ final boolean rebaseAllowed, final String analysisFingerprint, final VersionedProcessGroup mergedSnapshot) {
+ this.classifiedLocalChanges = Objects.requireNonNull(classifiedLocalChanges, "Classified local changes are required");
+ this.upstreamDifferences = Objects.requireNonNull(upstreamDifferences, "Upstream differences are required");
+ this.rebaseAllowed = rebaseAllowed;
+ this.analysisFingerprint = Objects.requireNonNull(analysisFingerprint, "Analysis fingerprint is required");
+ this.mergedSnapshot = mergedSnapshot;
+ }
+
+ public List<ClassifiedDifference> getClassifiedLocalChanges() {
+ return classifiedLocalChanges;
+ }
+
+ public Set<FlowDifference> getUpstreamDifferences() {
+ return upstreamDifferences;
+ }
+
+ public boolean isRebaseAllowed() {
+ return rebaseAllowed;
+ }
+
+ public String getAnalysisFingerprint() {
+ return analysisFingerprint;
+ }
+
+ public VersionedProcessGroup getMergedSnapshot() {
+ return mergedSnapshot;
+ }
+
+ public static class ClassifiedDifference {
+ private final FlowDifference difference;
+ private final RebaseClassification classification;
+ private final String conflictCode;
+ private final String conflictDetail;
+
+ public ClassifiedDifference(final FlowDifference difference, final RebaseClassification classification,
+ final String conflictCode, final String conflictDetail) {
+ this.difference = Objects.requireNonNull(difference, "Difference is required");
+ this.classification = Objects.requireNonNull(classification, "Classification is required");
+ this.conflictCode = conflictCode;
+ this.conflictDetail = conflictDetail;
+ }
+
+ public static ClassifiedDifference compatible(final FlowDifference difference) {
+ return new ClassifiedDifference(difference, RebaseClassification.COMPATIBLE, null, null);
+ }
+
+ public static ClassifiedDifference conflicting(final FlowDifference difference, final String conflictCode, final String conflictDetail) {
Review Comment:
Should the `conflictCode` be declared as an `enum` for reuse across the implementation?
##########
nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/VersionsResource.java:
##########
@@ -996,6 +998,186 @@ public Response deleteRevertRequest(
return deleteFlowUpdateRequest("revert-requests", revertRequestId, disconnectedNodeAcknowledged);
}
+ @GET
+ @Consumes(MediaType.WILDCARD)
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("rebase-analysis/process-groups/{id}")
+ @Operation(
+ summary = "Gets a Rebase Analysis for a Process Group",
+ responses = {
+ @ApiResponse(responseCode = "200", content = @Content(schema = @Schema(implementation = RebaseAnalysisEntity.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.")
+ },
+ description = "For a Process Group that is under Version Control, this will perform a rebase analysis by comparing "
+ + "local modifications against upstream changes between the current version and the specified target version. "
+ + "The analysis determines whether a rebase is allowed or if there are conflicts.",
+ security = {
+ @SecurityRequirement(name = "Read - /process-groups/{uuid}")
+ }
+ )
+ public Response getRebaseAnalysis(
+ @Parameter(description = "The process group id.") @PathParam("id") final String processGroupId,
+ @Parameter(description = "The target version to rebase to.", required = true) @QueryParam("targetVersion") final String targetVersion) {
+
+ if (targetVersion == null) {
+ throw new IllegalArgumentException("The target version must be specified.");
+ }
+
+ serviceFacade.authorizeAccess(lookup -> {
+ final ProcessGroupAuthorizable groupAuthorizable = lookup.getProcessGroup(processGroupId);
+ authorizeProcessGroup(groupAuthorizable, authorizer, lookup, RequestAction.READ, true,
+ false, false, false, true);
+ });
+
+ if (isReplicateRequest()) {
+ return replicate(HttpMethod.GET);
+ }
+
+ final RebaseAnalysisEntity entity = serviceFacade.getRebaseAnalysis(processGroupId, targetVersion);
+ return generateOkResponse(entity).build();
+ }
+
+ @POST
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("rebase-requests/process-groups/{id}")
+ @Operation(
+ summary = "Initiate a Rebase Request for a Process Group with the given ID",
+ responses = {
+ @ApiResponse(responseCode = "200", content = @Content(schema = @Schema(implementation = VersionedFlowUpdateRequestEntity.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.")
+ },
+ description = "For a Process Group that is already under Version Control, this will initiate the action of rebasing "
+ + "the flow to a different version while preserving compatible local changes. This can be a lengthy "
+ + "process, as it will stop any Processors and disable any Controller Services necessary to perform the action and then restart them. As a result, "
+ + "the endpoint will immediately return a VersionedFlowUpdateRequestEntity, and the process of rebasing the flow will occur "
+ + "asynchronously in the background. The client may then periodically poll the status of the request by issuing a GET request to "
+ + "/versions/rebase-requests/{requestId}. Once the request is completed, the client is expected to issue a DELETE request to "
+ + "/versions/rebase-requests/{requestId}. " + NON_GUARANTEED_ENDPOINT,
+ security = {
+ @SecurityRequirement(name = "Read - /process-groups/{uuid}"),
+ @SecurityRequirement(name = "Write - /process-groups/{uuid}"),
+ @SecurityRequirement(name = "Read - /{component-type}/{uuid} - For all encapsulated components"),
+ @SecurityRequirement(name = "Write - /{component-type}/{uuid} - For all encapsulated components"),
+ @SecurityRequirement(name = "Write - if the template contains any restricted components - /restricted-components"),
Review Comment:
This reference to restricted components is no longer applicable and should be removed
##########
nifi-registry/nifi-registry-core/nifi-registry-flow-diff/pom.xml:
##########
@@ -29,5 +29,9 @@
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-api</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
Review Comment:
Is this dependency needed if the ObjectMapper deep copy is removed?
##########
nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseEngine.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * 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.registry.flow.diff;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.flow.VersionedProcessGroup;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class RebaseEngine {
+
+ private static final ObjectMapper OBJECT_MAPPER = createObjectMapper();
+
+ private final Map<DifferenceType, RebaseHandler> handlerRegistry;
+
+ public RebaseEngine() {
+ this.handlerRegistry = new HashMap<>();
+ registerHandler(new PositionChangedRebaseHandler());
+ registerHandler(new SizeChangedRebaseHandler());
+ registerHandler(new BendpointsChangedRebaseHandler());
+ registerHandler(new PropertyChangedRebaseHandler());
+ registerHandler(new PropertyAddedRebaseHandler());
+ registerHandler(new CommentsChangedRebaseHandler());
+ }
+
+ public RebaseEngine(final Map<DifferenceType, RebaseHandler> handlerRegistry) {
+ this.handlerRegistry = new HashMap<>(handlerRegistry);
+ }
+
+ public RebaseAnalysis analyze(final Set<FlowDifference> localDifferences, final Set<FlowDifference> upstreamDifferences,
+ final VersionedProcessGroup targetSnapshot) {
+ final List<RebaseAnalysis.ClassifiedDifference> classifiedChanges = new ArrayList<>();
+ boolean allCompatible = true;
+
+ for (final FlowDifference localDifference : localDifferences) {
+ final RebaseHandler handler = handlerRegistry.get(localDifference.getDifferenceType());
+ if (handler == null) {
+ classifiedChanges.add(RebaseAnalysis.ClassifiedDifference.unsupported(localDifference, "NO_HANDLER",
+ "No rebase handler registered for difference type: " + localDifference.getDifferenceType().getDescription()));
+ allCompatible = false;
+ continue;
+ }
+
+ final RebaseAnalysis.ClassifiedDifference classified = handler.classify(localDifference, upstreamDifferences, targetSnapshot);
+ classifiedChanges.add(classified);
+ if (classified.getClassification() != RebaseClassification.COMPATIBLE) {
+ allCompatible = false;
+ }
+ }
+
+ VersionedProcessGroup mergedSnapshot = null;
+ if (allCompatible) {
+ mergedSnapshot = deepClone(targetSnapshot);
+ for (final RebaseAnalysis.ClassifiedDifference classified : classifiedChanges) {
+ final RebaseHandler handler = handlerRegistry.get(classified.getDifference().getDifferenceType());
+ if (handler != null) {
+ handler.apply(classified.getDifference(), mergedSnapshot);
+ }
+ }
+ }
+
+ final String fingerprint = computeAnalysisFingerprint(classifiedChanges, upstreamDifferences);
+ return new RebaseAnalysis(classifiedChanges, upstreamDifferences, allCompatible, fingerprint, mergedSnapshot);
+ }
+
+ public static String computeConflictKey(final FlowDifference difference) {
+ final DifferenceType type = difference.getDifferenceType();
+ final String componentId = resolveComponentIdentifier(difference);
+ final Optional<String> fieldName = difference.getFieldName();
+
+ return switch (type) {
+ case PROPERTY_CHANGED -> type.name() + ":" + componentId + ":" + fieldName.orElse("");
+ case POSITION_CHANGED, SIZE_CHANGED, COMMENTS_CHANGED, DESCRIPTION_CHANGED -> type.name() + ":" + componentId;
+ case BENDPOINTS_CHANGED -> type.name() + ":" + componentId;
+ default -> type.name() + ":" + componentId + ":" + fieldName.orElse("");
Review Comment:
The concatenation here is a bit hard to read, recommend defining one or more format strings and reusing
##########
nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseEngine.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * 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.registry.flow.diff;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.flow.VersionedProcessGroup;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class RebaseEngine {
+
+ private static final ObjectMapper OBJECT_MAPPER = createObjectMapper();
+
+ private final Map<DifferenceType, RebaseHandler> handlerRegistry;
+
+ public RebaseEngine() {
+ this.handlerRegistry = new HashMap<>();
+ registerHandler(new PositionChangedRebaseHandler());
+ registerHandler(new SizeChangedRebaseHandler());
+ registerHandler(new BendpointsChangedRebaseHandler());
+ registerHandler(new PropertyChangedRebaseHandler());
+ registerHandler(new PropertyAddedRebaseHandler());
+ registerHandler(new CommentsChangedRebaseHandler());
+ }
+
+ public RebaseEngine(final Map<DifferenceType, RebaseHandler> handlerRegistry) {
+ this.handlerRegistry = new HashMap<>(handlerRegistry);
+ }
+
+ public RebaseAnalysis analyze(final Set<FlowDifference> localDifferences, final Set<FlowDifference> upstreamDifferences,
+ final VersionedProcessGroup targetSnapshot) {
+ final List<RebaseAnalysis.ClassifiedDifference> classifiedChanges = new ArrayList<>();
+ boolean allCompatible = true;
+
+ for (final FlowDifference localDifference : localDifferences) {
+ final RebaseHandler handler = handlerRegistry.get(localDifference.getDifferenceType());
+ if (handler == null) {
+ classifiedChanges.add(RebaseAnalysis.ClassifiedDifference.unsupported(localDifference, "NO_HANDLER",
+ "No rebase handler registered for difference type: " + localDifference.getDifferenceType().getDescription()));
+ allCompatible = false;
+ continue;
+ }
+
+ final RebaseAnalysis.ClassifiedDifference classified = handler.classify(localDifference, upstreamDifferences, targetSnapshot);
+ classifiedChanges.add(classified);
+ if (classified.getClassification() != RebaseClassification.COMPATIBLE) {
+ allCompatible = false;
+ }
+ }
+
+ VersionedProcessGroup mergedSnapshot = null;
+ if (allCompatible) {
+ mergedSnapshot = deepClone(targetSnapshot);
+ for (final RebaseAnalysis.ClassifiedDifference classified : classifiedChanges) {
+ final RebaseHandler handler = handlerRegistry.get(classified.getDifference().getDifferenceType());
+ if (handler != null) {
+ handler.apply(classified.getDifference(), mergedSnapshot);
+ }
+ }
+ }
+
+ final String fingerprint = computeAnalysisFingerprint(classifiedChanges, upstreamDifferences);
+ return new RebaseAnalysis(classifiedChanges, upstreamDifferences, allCompatible, fingerprint, mergedSnapshot);
+ }
+
+ public static String computeConflictKey(final FlowDifference difference) {
+ final DifferenceType type = difference.getDifferenceType();
+ final String componentId = resolveComponentIdentifier(difference);
+ final Optional<String> fieldName = difference.getFieldName();
+
+ return switch (type) {
+ case PROPERTY_CHANGED -> type.name() + ":" + componentId + ":" + fieldName.orElse("");
+ case POSITION_CHANGED, SIZE_CHANGED, COMMENTS_CHANGED, DESCRIPTION_CHANGED -> type.name() + ":" + componentId;
+ case BENDPOINTS_CHANGED -> type.name() + ":" + componentId;
+ default -> type.name() + ":" + componentId + ":" + fieldName.orElse("");
+ };
+ }
+
+ VersionedProcessGroup deepClone(final VersionedProcessGroup source) {
+ try {
+ final byte[] serialized = OBJECT_MAPPER.writeValueAsBytes(source);
+ return OBJECT_MAPPER.readValue(serialized, VersionedProcessGroup.class);
Review Comment:
This seems like a bit of a hack to round trip and object through Jackson serialization. I recommend either explicit copy methods, or perhaps something like Spring BeanUtils
##########
nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseEngine.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * 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.registry.flow.diff;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.flow.VersionedProcessGroup;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class RebaseEngine {
+
+ private static final ObjectMapper OBJECT_MAPPER = createObjectMapper();
+
+ private final Map<DifferenceType, RebaseHandler> handlerRegistry;
+
+ public RebaseEngine() {
+ this.handlerRegistry = new HashMap<>();
+ registerHandler(new PositionChangedRebaseHandler());
+ registerHandler(new SizeChangedRebaseHandler());
+ registerHandler(new BendpointsChangedRebaseHandler());
+ registerHandler(new PropertyChangedRebaseHandler());
+ registerHandler(new PropertyAddedRebaseHandler());
+ registerHandler(new CommentsChangedRebaseHandler());
+ }
+
+ public RebaseEngine(final Map<DifferenceType, RebaseHandler> handlerRegistry) {
+ this.handlerRegistry = new HashMap<>(handlerRegistry);
+ }
+
+ public RebaseAnalysis analyze(final Set<FlowDifference> localDifferences, final Set<FlowDifference> upstreamDifferences,
+ final VersionedProcessGroup targetSnapshot) {
+ final List<RebaseAnalysis.ClassifiedDifference> classifiedChanges = new ArrayList<>();
+ boolean allCompatible = true;
+
+ for (final FlowDifference localDifference : localDifferences) {
+ final RebaseHandler handler = handlerRegistry.get(localDifference.getDifferenceType());
+ if (handler == null) {
+ classifiedChanges.add(RebaseAnalysis.ClassifiedDifference.unsupported(localDifference, "NO_HANDLER",
+ "No rebase handler registered for difference type: " + localDifference.getDifferenceType().getDescription()));
+ allCompatible = false;
+ continue;
+ }
+
+ final RebaseAnalysis.ClassifiedDifference classified = handler.classify(localDifference, upstreamDifferences, targetSnapshot);
+ classifiedChanges.add(classified);
+ if (classified.getClassification() != RebaseClassification.COMPATIBLE) {
+ allCompatible = false;
+ }
+ }
+
+ VersionedProcessGroup mergedSnapshot = null;
+ if (allCompatible) {
+ mergedSnapshot = deepClone(targetSnapshot);
+ for (final RebaseAnalysis.ClassifiedDifference classified : classifiedChanges) {
+ final RebaseHandler handler = handlerRegistry.get(classified.getDifference().getDifferenceType());
+ if (handler != null) {
+ handler.apply(classified.getDifference(), mergedSnapshot);
+ }
+ }
+ }
+
+ final String fingerprint = computeAnalysisFingerprint(classifiedChanges, upstreamDifferences);
+ return new RebaseAnalysis(classifiedChanges, upstreamDifferences, allCompatible, fingerprint, mergedSnapshot);
+ }
+
+ public static String computeConflictKey(final FlowDifference difference) {
+ final DifferenceType type = difference.getDifferenceType();
+ final String componentId = resolveComponentIdentifier(difference);
+ final Optional<String> fieldName = difference.getFieldName();
+
+ return switch (type) {
+ case PROPERTY_CHANGED -> type.name() + ":" + componentId + ":" + fieldName.orElse("");
+ case POSITION_CHANGED, SIZE_CHANGED, COMMENTS_CHANGED, DESCRIPTION_CHANGED -> type.name() + ":" + componentId;
+ case BENDPOINTS_CHANGED -> type.name() + ":" + componentId;
+ default -> type.name() + ":" + componentId + ":" + fieldName.orElse("");
+ };
+ }
+
+ VersionedProcessGroup deepClone(final VersionedProcessGroup source) {
+ try {
+ final byte[] serialized = OBJECT_MAPPER.writeValueAsBytes(source);
+ return OBJECT_MAPPER.readValue(serialized, VersionedProcessGroup.class);
+ } catch (final Exception e) {
+ throw new RuntimeException("Failed to deep clone VersionedProcessGroup", e);
+ }
+ }
+
+ private static String resolveComponentIdentifier(final FlowDifference difference) {
+ if (difference.getComponentB() != null) {
+ return difference.getComponentB().getIdentifier();
+ }
+ if (difference.getComponentA() != null) {
+ return difference.getComponentA().getIdentifier();
+ }
+ return "";
+ }
+
+ private String computeAnalysisFingerprint(final List<RebaseAnalysis.ClassifiedDifference> classifiedChanges,
+ final Set<FlowDifference> upstreamDifferences) {
+ final SortedSet<String> sortedKeys = new TreeSet<>();
+
+ for (final RebaseAnalysis.ClassifiedDifference classified : classifiedChanges) {
+ final String key = computeConflictKey(classified.getDifference());
+ sortedKeys.add("local:" + key + ":" + classified.getClassification().name());
Review Comment:
See other notes regarding standard formatting for keys
##########
nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/TestPropertyChangedRebaseHandler.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.registry.flow.diff;
+
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
+import org.apache.nifi.flow.VersionedPropertyDescriptor;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+public class TestPropertyChangedRebaseHandler {
Review Comment:
Since these are net-new classes, recommend using `Test` as the suffix instead of the prefix for the class. The public modifiers can also be removed
##########
nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/TestPropertyChangedRebaseHandler.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.registry.flow.diff;
+
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
+import org.apache.nifi.flow.VersionedPropertyDescriptor;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+public class TestPropertyChangedRebaseHandler {
+
+ private PropertyChangedRebaseHandler handler;
+
+ @BeforeEach
+ public void setup() {
+ handler = new PropertyChangedRebaseHandler();
+ }
+
+ @Test
+ public void testNoUpstreamConflict() {
+ final VersionedProcessor processor = createProcessorWithProperty("proc-a", "propX", "oldVal");
Review Comment:
There are a number of repeated string values in this test that should be moved to static final variables
##########
nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/VersionsResource.java:
##########
@@ -1237,8 +1419,12 @@ protected ProcessGroupEntity performUpdateFlow(final String groupId, final Revis
versionControlInfo.setVersion(metadata.getVersion());
versionControlInfo.setState(flowSnapshot.isLatest() ? VersionedFlowState.UP_TO_DATE.name() : VersionedFlowState.STALE.name());
- return serviceFacade.updateProcessGroupContents(revision, groupId, versionControlInfo, flowSnapshot, idGenerationSeed,
+ final ProcessGroupEntity result = serviceFacade.updateProcessGroupContents(revision, groupId, versionControlInfo, flowSnapshot, idGenerationSeed,
verifyNotModified, false, updateDescendantVersionedFlows);
+
+ serviceFacade.resetVersionControlSnapshotAfterRebase(groupId);
Review Comment:
It seems like a failure to update the Process Group contents would fail to reset the snapshot, leaving the snapshot entry in the Map within the Standard Service Facade. Perhaps putting this in a `finally` block would be sufficient?
##########
nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java:
##########
@@ -468,6 +473,8 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
private static final int VALIDATION_WAIT_MILLIS = 50;
private static final String ROOT_PROCESS_GROUP = "RootProcessGroup";
+ private final Map<String, VersionedProcessGroup> rebaseCleanSnapshots = new ConcurrentHashMap<>();
Review Comment:
Are there any concerns about concurrent rebase requests for the same Process Group?
##########
nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseEngine.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * 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.registry.flow.diff;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.flow.VersionedProcessGroup;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class RebaseEngine {
+
+ private static final ObjectMapper OBJECT_MAPPER = createObjectMapper();
+
+ private final Map<DifferenceType, RebaseHandler> handlerRegistry;
+
+ public RebaseEngine() {
+ this.handlerRegistry = new HashMap<>();
+ registerHandler(new PositionChangedRebaseHandler());
+ registerHandler(new SizeChangedRebaseHandler());
+ registerHandler(new BendpointsChangedRebaseHandler());
+ registerHandler(new PropertyChangedRebaseHandler());
+ registerHandler(new PropertyAddedRebaseHandler());
+ registerHandler(new CommentsChangedRebaseHandler());
+ }
+
+ public RebaseEngine(final Map<DifferenceType, RebaseHandler> handlerRegistry) {
+ this.handlerRegistry = new HashMap<>(handlerRegistry);
+ }
+
+ public RebaseAnalysis analyze(final Set<FlowDifference> localDifferences, final Set<FlowDifference> upstreamDifferences,
+ final VersionedProcessGroup targetSnapshot) {
+ final List<RebaseAnalysis.ClassifiedDifference> classifiedChanges = new ArrayList<>();
+ boolean allCompatible = true;
+
+ for (final FlowDifference localDifference : localDifferences) {
+ final RebaseHandler handler = handlerRegistry.get(localDifference.getDifferenceType());
+ if (handler == null) {
+ classifiedChanges.add(RebaseAnalysis.ClassifiedDifference.unsupported(localDifference, "NO_HANDLER",
+ "No rebase handler registered for difference type: " + localDifference.getDifferenceType().getDescription()));
+ allCompatible = false;
+ continue;
+ }
+
+ final RebaseAnalysis.ClassifiedDifference classified = handler.classify(localDifference, upstreamDifferences, targetSnapshot);
+ classifiedChanges.add(classified);
+ if (classified.getClassification() != RebaseClassification.COMPATIBLE) {
+ allCompatible = false;
+ }
+ }
+
+ VersionedProcessGroup mergedSnapshot = null;
+ if (allCompatible) {
+ mergedSnapshot = deepClone(targetSnapshot);
+ for (final RebaseAnalysis.ClassifiedDifference classified : classifiedChanges) {
+ final RebaseHandler handler = handlerRegistry.get(classified.getDifference().getDifferenceType());
+ if (handler != null) {
+ handler.apply(classified.getDifference(), mergedSnapshot);
+ }
+ }
+ }
+
+ final String fingerprint = computeAnalysisFingerprint(classifiedChanges, upstreamDifferences);
+ return new RebaseAnalysis(classifiedChanges, upstreamDifferences, allCompatible, fingerprint, mergedSnapshot);
+ }
+
+ public static String computeConflictKey(final FlowDifference difference) {
+ final DifferenceType type = difference.getDifferenceType();
+ final String componentId = resolveComponentIdentifier(difference);
+ final Optional<String> fieldName = difference.getFieldName();
+
+ return switch (type) {
+ case PROPERTY_CHANGED -> type.name() + ":" + componentId + ":" + fieldName.orElse("");
+ case POSITION_CHANGED, SIZE_CHANGED, COMMENTS_CHANGED, DESCRIPTION_CHANGED -> type.name() + ":" + componentId;
+ case BENDPOINTS_CHANGED -> type.name() + ":" + componentId;
+ default -> type.name() + ":" + componentId + ":" + fieldName.orElse("");
+ };
+ }
+
+ VersionedProcessGroup deepClone(final VersionedProcessGroup source) {
+ try {
+ final byte[] serialized = OBJECT_MAPPER.writeValueAsBytes(source);
+ return OBJECT_MAPPER.readValue(serialized, VersionedProcessGroup.class);
+ } catch (final Exception e) {
+ throw new RuntimeException("Failed to deep clone VersionedProcessGroup", e);
+ }
+ }
+
+ private static String resolveComponentIdentifier(final FlowDifference difference) {
+ if (difference.getComponentB() != null) {
+ return difference.getComponentB().getIdentifier();
+ }
+ if (difference.getComponentA() != null) {
+ return difference.getComponentA().getIdentifier();
+ }
+ return "";
Review Comment:
Is empty string a valid return?
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org