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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
