This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 6b1116e4b7f NIFI-16235 Support rebasing locally added Controller
Services in versioned Process Groups (#11571)
6b1116e4b7f is described below
commit 6b1116e4b7f087b91143d3a1fe1118f50579b617
Author: Pierre Villard <[email protected]>
AuthorDate: Mon Sep 21 15:37:41 2026 +0200
NIFI-16235 Support rebasing locally added Controller Services in versioned
Process Groups (#11571)
Signed-off-by: David Handermann <[email protected]>
---
.../flow/diff/ComponentAddedRebaseHandler.java | 124 +++++++++
.../nifi/registry/flow/diff/RebaseAnalysis.java | 15 ++
.../registry/flow/diff/RebaseConflictCode.java | 10 +
.../nifi/registry/flow/diff/RebaseHandler.java | 4 +
.../registry/flow/diff/RebaseHandlerUtils.java | 15 ++
.../registry/flow/diff/StandardRebaseEngine.java | 3 +-
.../flow/diff/ComponentAddedRebaseHandlerTest.java | 296 +++++++++++++++++++++
.../nifi/registry/flow/diff/RebaseEngineTest.java | 266 +++++++++++++++++-
.../FakeDynamicPropertiesControllerService.java | 46 ++++
.../org.apache.nifi.controller.ControllerService | 1 +
.../tests/system/registry/RebaseVersionIT.java | 218 +++++++++++++--
11 files changed, 975 insertions(+), 23 deletions(-)
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/ComponentAddedRebaseHandler.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/ComponentAddedRebaseHandler.java
new file mode 100644
index 00000000000..4fde24f6741
--- /dev/null
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/ComponentAddedRebaseHandler.java
@@ -0,0 +1,124 @@
+/*
+ * 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.VersionedComponent;
+import org.apache.nifi.flow.VersionedControllerService;
+import org.apache.nifi.flow.VersionedProcessGroup;
+
+import java.util.HashSet;
+import java.util.Set;
+
+public class ComponentAddedRebaseHandler implements RebaseHandler {
+
+ private static final String NULL_COMPONENT_TYPE = "null";
+
+ @Override
+ public DifferenceType getSupportedType() {
+ return DifferenceType.COMPONENT_ADDED;
+ }
+
+ @Override
+ public RebaseAnalysis.ClassifiedDifference classify(final FlowDifference
localDifference, final Set<FlowDifference> upstreamDifferences,
+ final
VersionedProcessGroup targetSnapshot) {
+ final VersionedComponent addedComponent =
localDifference.getComponentB();
+ if (!(addedComponent instanceof VersionedControllerService
controllerService)) {
+ final String componentType = addedComponent == null ?
NULL_COMPONENT_TYPE : addedComponent.getClass().getSimpleName();
+ return
RebaseAnalysis.ClassifiedDifference.unsupported(localDifference,
RebaseConflictCode.UNSUPPORTED_COMPONENT_TYPE,
+ "Local component addition type %s is not supported for
rebase".formatted(componentType));
+ }
+
+ final String parentGroupIdentifier =
controllerService.getGroupIdentifier();
+ if (parentGroupIdentifier == null) {
+ return
RebaseAnalysis.ClassifiedDifference.unsupported(localDifference,
RebaseConflictCode.COMPONENT_NOT_FOUND,
+ "Controller Service %s does not specify a parent Process
Group".formatted(controllerService.getIdentifier()));
+ }
+
+ final VersionedProcessGroup parentGroup =
resolveParentGroup(targetSnapshot, parentGroupIdentifier, upstreamDifferences);
+ if (parentGroup == null) {
+ return
RebaseAnalysis.ClassifiedDifference.unsupported(localDifference,
RebaseConflictCode.COMPONENT_NOT_FOUND,
+ "Parent Process Group %s for Controller Service %s not
found in target snapshot"
+ .formatted(parentGroupIdentifier,
controllerService.getIdentifier()));
+ }
+
+ final VersionedComponent collidingComponent =
RebaseHandlerUtils.findComponentById(targetSnapshot,
controllerService.getIdentifier());
+ if (collidingComponent != null) {
+ return
RebaseAnalysis.ClassifiedDifference.conflicting(localDifference,
RebaseConflictCode.COMPONENT_IDENTIFIER_COLLISION,
+ "Target snapshot already contains component %s with
identifier %s"
+
.formatted(collidingComponent.getClass().getSimpleName(),
controllerService.getIdentifier()));
+ }
+
+ return RebaseAnalysis.ClassifiedDifference.compatible(localDifference,
parentGroup.getIdentifier());
+ }
+
+ @Override
+ public void apply(final FlowDifference localDifference, final
VersionedProcessGroup mergedFlow) {
+ final VersionedControllerService controllerService =
(VersionedControllerService) localDifference.getComponentB();
+ final VersionedProcessGroup parentGroup =
RebaseHandlerUtils.findProcessGroupById(mergedFlow,
controllerService.getGroupIdentifier());
+ apply(localDifference, mergedFlow, parentGroup,
controllerService.getGroupIdentifier());
+ }
+
+ @Override
+ public void apply(final RebaseAnalysis.ClassifiedDifference
classifiedDifference, final VersionedProcessGroup mergedFlow) {
+ final FlowDifference localDifference =
classifiedDifference.getDifference();
+ final String parentGroupIdentifier = (String)
classifiedDifference.getContext();
+ final VersionedProcessGroup parentGroup =
RebaseHandlerUtils.findProcessGroupById(mergedFlow, parentGroupIdentifier);
+ apply(localDifference, mergedFlow, parentGroup, parentGroupIdentifier);
+ }
+
+ private void apply(final FlowDifference localDifference, final
VersionedProcessGroup mergedFlow, final VersionedProcessGroup parentGroup,
+ final String parentGroupIdentifier) {
+ final VersionedControllerService controllerService =
(VersionedControllerService) localDifference.getComponentB();
+ if (parentGroup == null) {
+ throw new IllegalStateException("Parent Process Group %s for
Controller Service %s was verified during classification but is absent during
apply"
+ .formatted(parentGroupIdentifier,
controllerService.getIdentifier()));
+ }
+
+ controllerService.setGroupIdentifier(parentGroup.getIdentifier());
+
+ final VersionedComponent existingComponent =
RebaseHandlerUtils.findComponentById(mergedFlow,
controllerService.getIdentifier());
+ if (existingComponent != null) {
+ throw new IllegalStateException("Merged flow already contains
component %s with identifier %s"
+ .formatted(existingComponent.getClass().getSimpleName(),
controllerService.getIdentifier()));
+ }
+
+ final Set<VersionedControllerService> controllerServices =
parentGroup.getControllerServices();
+ if (controllerServices == null) {
+ parentGroup.setControllerServices(new HashSet<>());
+ }
+ parentGroup.getControllerServices().add(controllerService);
+ }
+
+ private VersionedProcessGroup resolveParentGroup(final
VersionedProcessGroup targetSnapshot, final String parentGroupIdentifier,
+ final Set<FlowDifference>
upstreamDifferences) {
+ final VersionedProcessGroup parentGroup =
RebaseHandlerUtils.findProcessGroupById(targetSnapshot, parentGroupIdentifier);
+ if (parentGroup != null) {
+ return parentGroup;
+ }
+
+ final boolean parentRemoved = upstreamDifferences.stream()
+ .filter(difference -> difference.getDifferenceType() ==
DifferenceType.COMPONENT_REMOVED)
+ .map(FlowDifference::getComponentA)
+ .filter(VersionedProcessGroup.class::isInstance)
+ .map(VersionedComponent::getIdentifier)
+ .anyMatch(parentGroupIdentifier::equals);
+
+ return parentRemoved ? null : targetSnapshot;
+ }
+
+}
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseAnalysis.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseAnalysis.java
index b5418139e39..0e0c4907aee 100644
---
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseAnalysis.java
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseAnalysis.java
@@ -65,19 +65,30 @@ public class RebaseAnalysis {
private final RebaseClassification classification;
private final RebaseConflictCode conflictCode;
private final String conflictDetail;
+ private final Object context;
public ClassifiedDifference(final FlowDifference difference, final
RebaseClassification classification,
final RebaseConflictCode conflictCode,
final String conflictDetail) {
+ this(difference, classification, conflictCode, conflictDetail,
null);
+ }
+
+ public ClassifiedDifference(final FlowDifference difference, final
RebaseClassification classification,
+ final RebaseConflictCode conflictCode,
final String conflictDetail, final Object context) {
this.difference = Objects.requireNonNull(difference, "Difference
is required");
this.classification = Objects.requireNonNull(classification,
"Classification is required");
this.conflictCode = conflictCode;
this.conflictDetail = conflictDetail;
+ this.context = context;
}
public static ClassifiedDifference compatible(final FlowDifference
difference) {
return new ClassifiedDifference(difference,
RebaseClassification.COMPATIBLE, null, null);
}
+ public static ClassifiedDifference compatible(final FlowDifference
difference, final Object context) {
+ return new ClassifiedDifference(difference,
RebaseClassification.COMPATIBLE, null, null, context);
+ }
+
public static ClassifiedDifference conflicting(final FlowDifference
difference, final RebaseConflictCode conflictCode, final String conflictDetail)
{
return new ClassifiedDifference(difference,
RebaseClassification.CONFLICTING, conflictCode, conflictDetail);
}
@@ -101,5 +112,9 @@ public class RebaseAnalysis {
public String getConflictDetail() {
return conflictDetail;
}
+
+ public Object getContext() {
+ return context;
+ }
}
}
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseConflictCode.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseConflictCode.java
index 311ced78b33..ad651b6d017 100644
---
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseConflictCode.java
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseConflictCode.java
@@ -32,6 +32,11 @@ public enum RebaseConflictCode {
*/
MISSING_FIELD_NAME,
+ /**
+ * The registered handler does not support the local change's component
type.
+ */
+ UNSUPPORTED_COMPONENT_TYPE,
+
/**
* Both the local and upstream flows modified the same property on the
same component.
*/
@@ -47,6 +52,11 @@ public enum RebaseConflictCode {
*/
COMPONENT_NOT_FOUND,
+ /**
+ * The target version already contains a component with the same
identifier as the local addition.
+ */
+ COMPONENT_IDENTIFIER_COLLISION,
+
/**
* The property descriptor targeted by the local change changed in an
incompatible way in the target version.
*/
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseHandler.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseHandler.java
index acb0b3d4723..431bc9c6982 100644
---
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseHandler.java
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseHandler.java
@@ -28,4 +28,8 @@ public interface RebaseHandler {
RebaseAnalysis.ClassifiedDifference classify(FlowDifference
localDifference, Set<FlowDifference> upstreamDifferences, VersionedProcessGroup
targetSnapshot);
void apply(FlowDifference localDifference, VersionedProcessGroup
mergedFlow);
+
+ default void apply(final RebaseAnalysis.ClassifiedDifference
classifiedDifference, final VersionedProcessGroup mergedFlow) {
+ apply(classifiedDifference.getDifference(), mergedFlow);
+ }
}
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseHandlerUtils.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseHandlerUtils.java
index f870ea4b8a8..c1d2702d5c6 100644
---
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseHandlerUtils.java
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseHandlerUtils.java
@@ -89,6 +89,21 @@ class RebaseHandlerUtils {
return null;
}
+ static VersionedProcessGroup findProcessGroupById(final
VersionedProcessGroup group, final String identifier) {
+ if (identifier.equals(group.getIdentifier()) ||
identifier.equals(group.getInstanceIdentifier())) {
+ return group;
+ }
+
+ for (final VersionedProcessGroup childGroup :
group.getProcessGroups()) {
+ final VersionedProcessGroup result =
findProcessGroupById(childGroup, identifier);
+ if (result != null) {
+ return result;
+ }
+ }
+
+ return null;
+ }
+
static VersionedConnection findConnectionById(final VersionedProcessGroup
group, final String identifier) {
for (final VersionedConnection connection : group.getConnections()) {
if (identifier.equals(connection.getIdentifier())) {
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardRebaseEngine.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardRebaseEngine.java
index b942d2bfccf..57153781cdf 100644
---
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardRebaseEngine.java
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardRebaseEngine.java
@@ -46,6 +46,7 @@ public class StandardRebaseEngine implements RebaseEngine {
registerHandler(new PositionChangedRebaseHandler());
registerHandler(new SizeChangedRebaseHandler());
registerHandler(new BendpointsChangedRebaseHandler());
+ registerHandler(new ComponentAddedRebaseHandler());
registerHandler(new PropertyChangedRebaseHandler());
registerHandler(new PropertyAddedRebaseHandler());
registerHandler(new CommentsChangedRebaseHandler());
@@ -76,7 +77,7 @@ public class StandardRebaseEngine implements RebaseEngine {
for (final RebaseAnalysis.ClassifiedDifference classified :
classifiedChanges) {
final RebaseHandler handler =
handlerRegistry.get(classified.getDifference().getDifferenceType());
if (handler != null) {
- handler.apply(classified.getDifference(), mergedSnapshot);
+ handler.apply(classified, mergedSnapshot);
}
}
}
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/ComponentAddedRebaseHandlerTest.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/ComponentAddedRebaseHandlerTest.java
new file mode 100644
index 00000000000..06a2de47b08
--- /dev/null
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/ComponentAddedRebaseHandlerTest.java
@@ -0,0 +1,296 @@
+/*
+ * 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.Bundle;
+import org.apache.nifi.flow.ScheduledState;
+import org.apache.nifi.flow.VersionedControllerService;
+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;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class ComponentAddedRebaseHandlerTest {
+
+ private static final String ROOT = "root";
+ private static final String TARGET_ROOT = "target-root";
+ private static final String CHILD = "child";
+ private static final String PROCESSOR_ID = "processor-a";
+ private static final String EXISTING_SERVICE_ID = "service-x";
+ private static final String ADDED_SERVICE_ID = "service-y";
+ private static final String SERVICE_REFERENCE_PROPERTY =
"delegate.service";
+ private static final String LOCAL_SERVICE_NAME = "Local Controller
Service";
+ private static final String LOCAL_SERVICE_TYPE =
"org.apache.nifi.services.LocalControllerService";
+ private static final String BUNDLE_GROUP = "group";
+ private static final String BUNDLE_ARTIFACT = "artifact";
+ private static final String BUNDLE_VERSION = "1.0.0";
+ private static final String LOCAL_COMMENTS = "local comments";
+ private static final String SERVICE_ENABLED_PROPERTY = "service.enabled";
+ private static final String SERVICE_ENABLED_VALUE = "true";
+
+ private ComponentAddedRebaseHandler handler;
+
+ @BeforeEach
+ void setup() {
+ handler = new ComponentAddedRebaseHandler();
+ }
+
+ @Test
+ void testClassifyReferencedControllerServiceAdditionIsCompatible() {
+ final VersionedControllerService addedService =
createControllerService(ADDED_SERVICE_ID, ROOT);
+ addedService.setProperties(Map.of(SERVICE_REFERENCE_PROPERTY,
EXISTING_SERVICE_ID));
+ addedService.setPropertyDescriptors(Map.of(SERVICE_REFERENCE_PROPERTY,
createDescriptor(SERVICE_REFERENCE_PROPERTY, false, false)));
+
+ final FlowDifference localDifference = new
StandardFlowDifference(DifferenceType.COMPONENT_ADDED, null, addedService,
+ null, addedService, "Referenced controller service added
locally");
+
+ final VersionedProcessGroup targetSnapshot =
createTargetSnapshotWithExistingService(EXISTING_SERVICE_ID, ROOT);
+ targetSnapshot.setIdentifier(TARGET_ROOT);
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, Collections.emptySet(), targetSnapshot);
+
+ assertEquals(RebaseClassification.COMPATIBLE,
result.getClassification());
+ assertEquals(ROOT, addedService.getGroupIdentifier());
+ }
+
+ @Test
+ void testClassifyUnreferencedControllerServiceAdditionIsCompatible() {
+ final VersionedControllerService addedService =
createControllerService(ADDED_SERVICE_ID, ROOT);
+ final FlowDifference localDifference = new
StandardFlowDifference(DifferenceType.COMPONENT_ADDED, null, addedService,
+ null, addedService, "Unreferenced controller service added
locally");
+
+ final VersionedProcessGroup targetSnapshot =
createTargetSnapshotWithExistingService(EXISTING_SERVICE_ID, ROOT);
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, Collections.emptySet(), targetSnapshot);
+
+ assertEquals(RebaseClassification.COMPATIBLE,
result.getClassification());
+ }
+
+ @Test
+ void testClassifyNonControllerServiceAdditionIsUnsupported() {
+ final VersionedProcessor processor = new VersionedProcessor();
+ processor.setIdentifier(PROCESSOR_ID);
+
+ final FlowDifference localDifference = new
StandardFlowDifference(DifferenceType.COMPONENT_ADDED, null, processor,
+ null, processor, "Processor added locally");
+
+ final VersionedProcessGroup targetSnapshot = createRootGroup();
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, Collections.emptySet(), targetSnapshot);
+
+ assertEquals(RebaseClassification.UNSUPPORTED,
result.getClassification());
+ assertEquals(RebaseConflictCode.UNSUPPORTED_COMPONENT_TYPE,
result.getConflictCode());
+ }
+
+ @Test
+ void testClassifyNullParentIsUnsupported() {
+ final VersionedControllerService addedService =
createControllerService(ADDED_SERVICE_ID, null);
+ final FlowDifference localDifference = new
StandardFlowDifference(DifferenceType.COMPONENT_ADDED, null, addedService,
+ null, addedService, "Controller service added without parent");
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, Collections.emptySet(), createRootGroup());
+
+ assertEquals(RebaseClassification.UNSUPPORTED,
result.getClassification());
+ assertEquals(RebaseConflictCode.COMPONENT_NOT_FOUND,
result.getConflictCode());
+ }
+
+ @Test
+ void testClassifyMissingParentIsUnsupported() {
+ final VersionedControllerService addedService =
createControllerService(ADDED_SERVICE_ID, CHILD);
+ final FlowDifference localDifference = new
StandardFlowDifference(DifferenceType.COMPONENT_ADDED, null, addedService,
+ null, addedService, "Controller service added to missing
parent");
+ final VersionedProcessGroup removedParent = createChildGroup(CHILD);
+ final Set<FlowDifference> upstreamDifferences = Set.of(new
StandardFlowDifference(DifferenceType.COMPONENT_REMOVED, removedParent, null,
+ removedParent, null, "Parent removed upstream"));
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, upstreamDifferences, createRootGroup());
+
+ assertEquals(RebaseClassification.UNSUPPORTED,
result.getClassification());
+ assertEquals(RebaseConflictCode.COMPONENT_NOT_FOUND,
result.getConflictCode());
+ }
+
+ @Test
+ void testClassifySameIdentifierTargetCollisionIsConflicting() {
+ final VersionedControllerService addedService =
createControllerService(ADDED_SERVICE_ID, ROOT);
+ final FlowDifference localDifference = new
StandardFlowDifference(DifferenceType.COMPONENT_ADDED, null, addedService,
+ null, addedService, "Controller service added with colliding
identifier");
+
+ final VersionedProcessGroup childGroup = new VersionedProcessGroup();
+ childGroup.setIdentifier(CHILD);
+
+ final VersionedProcessor collidingProcessor = new VersionedProcessor();
+ collidingProcessor.setIdentifier(ADDED_SERVICE_ID);
+ childGroup.getProcessors().add(collidingProcessor);
+
+ final VersionedProcessGroup targetSnapshot = createRootGroup();
+ targetSnapshot.getProcessGroups().add(childGroup);
+
+ final RebaseAnalysis.ClassifiedDifference result =
handler.classify(localDifference, Collections.emptySet(), targetSnapshot);
+
+ assertEquals(RebaseClassification.CONFLICTING,
result.getClassification());
+ assertEquals(RebaseConflictCode.COMPONENT_IDENTIFIER_COLLISION,
result.getConflictCode());
+ }
+
+ @Test
+ void
testApplyAddsControllerServiceToRootPreservingIdentityAndConfiguration() {
+ final VersionedControllerService addedService =
createControllerService(ADDED_SERVICE_ID, ROOT);
+ final FlowDifference localDifference = new
StandardFlowDifference(DifferenceType.COMPONENT_ADDED, null, addedService,
+ null, addedService, "Controller service added to root");
+
+ final VersionedProcessGroup mergedFlow = new VersionedProcessGroup();
+ mergedFlow.setIdentifier(TARGET_ROOT);
+ mergedFlow.setInstanceIdentifier(ROOT);
+
+ handler.apply(localDifference, mergedFlow);
+
+ assertNotNull(mergedFlow.getControllerServices());
+ assertEquals(1, mergedFlow.getControllerServices().size());
+
+ final VersionedControllerService insertedService =
mergedFlow.getControllerServices().iterator().next();
+ assertSame(addedService, insertedService);
+ assertEquals(ADDED_SERVICE_ID, insertedService.getIdentifier());
+ assertEquals(TARGET_ROOT, insertedService.getGroupIdentifier());
+ assertEquals(LOCAL_SERVICE_NAME, insertedService.getName());
+ assertEquals(LOCAL_SERVICE_TYPE, insertedService.getType());
+ assertEquals(BUNDLE_GROUP, insertedService.getBundle().getGroup());
+ assertEquals(BUNDLE_ARTIFACT,
insertedService.getBundle().getArtifact());
+ assertEquals(BUNDLE_VERSION, insertedService.getBundle().getVersion());
+ assertSame(addedService.getProperties(),
insertedService.getProperties());
+ assertSame(addedService.getPropertyDescriptors(),
insertedService.getPropertyDescriptors());
+ assertEquals(LOCAL_COMMENTS, insertedService.getComments());
+ assertEquals(ScheduledState.DISABLED,
insertedService.getScheduledState());
+ }
+
+ @Test
+ void testApplyAddsControllerServiceToNestedParent() {
+ final VersionedControllerService addedService =
createControllerService(ADDED_SERVICE_ID, CHILD);
+ final FlowDifference localDifference = new
StandardFlowDifference(DifferenceType.COMPONENT_ADDED, null, addedService,
+ null, addedService, "Controller service added to nested
group");
+
+ final VersionedProcessGroup childGroup = new VersionedProcessGroup();
+ childGroup.setIdentifier(CHILD);
+
+ final VersionedProcessGroup mergedFlow = createRootGroup();
+ mergedFlow.getProcessGroups().add(childGroup);
+
+ handler.apply(localDifference, mergedFlow);
+
+ assertEquals(1, childGroup.getControllerServices().size());
+ assertSame(addedService,
childGroup.getControllerServices().iterator().next());
+ assertEquals(0, mergedFlow.getControllerServices().size());
+ }
+
+ @Test
+ void testApplyThrowsWhenVerifiedParentIsMissing() {
+ final VersionedControllerService addedService =
createControllerService(ADDED_SERVICE_ID, CHILD);
+ final FlowDifference localDifference = new
StandardFlowDifference(DifferenceType.COMPONENT_ADDED, null, addedService,
+ null, addedService, "Controller service added to missing
parent");
+
+ final IllegalStateException exception =
assertThrows(IllegalStateException.class,
+ () -> handler.apply(localDifference, createRootGroup()));
+
+ assertEquals("Parent Process Group child for Controller Service
service-y was verified during classification but is absent during apply",
+ exception.getMessage());
+ }
+
+ @Test
+ void testApplyThrowsWhenIdentifierAlreadyExists() {
+ final VersionedControllerService addedService =
createControllerService(ADDED_SERVICE_ID, ROOT);
+ final FlowDifference localDifference = new
StandardFlowDifference(DifferenceType.COMPONENT_ADDED, null, addedService,
+ null, addedService, "Controller service added with colliding
identifier");
+
+ final VersionedProcessor existingProcessor = new VersionedProcessor();
+ existingProcessor.setIdentifier(ADDED_SERVICE_ID);
+
+ final VersionedProcessGroup mergedFlow = createRootGroup();
+ mergedFlow.getProcessors().add(existingProcessor);
+
+ final IllegalStateException exception =
assertThrows(IllegalStateException.class,
+ () -> handler.apply(localDifference, mergedFlow));
+
+ assertEquals("Merged flow already contains component
VersionedProcessor with identifier service-y", exception.getMessage());
+ }
+
+ private VersionedProcessGroup
createTargetSnapshotWithExistingService(final String serviceIdentifier, final
String groupIdentifier) {
+ final VersionedProcessGroup rootGroup = createRootGroup();
+ final VersionedProcessGroup parentGroup = ROOT.equals(groupIdentifier)
? rootGroup : createChildGroup(groupIdentifier);
+
+ if (parentGroup != rootGroup) {
+ rootGroup.getProcessGroups().add(parentGroup);
+ }
+
+
parentGroup.getControllerServices().add(createControllerService(serviceIdentifier,
groupIdentifier));
+ return rootGroup;
+ }
+
+ private VersionedProcessGroup createRootGroup() {
+ final VersionedProcessGroup rootGroup = new VersionedProcessGroup();
+ rootGroup.setIdentifier(ROOT);
+ rootGroup.setInstanceIdentifier(ROOT);
+ rootGroup.setControllerServices(new HashSet<>());
+ return rootGroup;
+ }
+
+ private VersionedProcessGroup createChildGroup(final String identifier) {
+ final VersionedProcessGroup childGroup = new VersionedProcessGroup();
+ childGroup.setIdentifier(identifier);
+ childGroup.setInstanceIdentifier(identifier);
+ childGroup.setControllerServices(new HashSet<>());
+ return childGroup;
+ }
+
+ private VersionedControllerService createControllerService(final String
identifier, final String groupIdentifier) {
+ final VersionedControllerService service = new
VersionedControllerService();
+ service.setIdentifier(identifier);
+ service.setGroupIdentifier(groupIdentifier);
+ service.setName(LOCAL_SERVICE_NAME);
+ service.setType(LOCAL_SERVICE_TYPE);
+ service.setBundle(new Bundle(BUNDLE_GROUP, BUNDLE_ARTIFACT,
BUNDLE_VERSION));
+ service.setScheduledState(ScheduledState.DISABLED);
+ service.setComments(LOCAL_COMMENTS);
+
+ final Map<String, String> properties = new HashMap<>();
+ properties.put(SERVICE_ENABLED_PROPERTY, SERVICE_ENABLED_VALUE);
+ service.setProperties(properties);
+
+ final Map<String, VersionedPropertyDescriptor> descriptors = new
HashMap<>();
+ descriptors.put(SERVICE_ENABLED_PROPERTY,
createDescriptor(SERVICE_ENABLED_PROPERTY, false, false));
+ service.setPropertyDescriptors(descriptors);
+ return service;
+ }
+
+ private VersionedPropertyDescriptor createDescriptor(final String name,
final boolean dynamic, final boolean sensitive) {
+ final VersionedPropertyDescriptor descriptor = new
VersionedPropertyDescriptor();
+ descriptor.setName(name);
+ descriptor.setDynamic(dynamic);
+ descriptor.setSensitive(sensitive);
+ return descriptor;
+ }
+}
diff --git
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/RebaseEngineTest.java
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/RebaseEngineTest.java
index 74e92993272..66b9287b3a6 100644
---
a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/RebaseEngineTest.java
+++
b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/RebaseEngineTest.java
@@ -17,8 +17,11 @@
package org.apache.nifi.registry.flow.diff;
+import org.apache.nifi.flow.Bundle;
import org.apache.nifi.flow.Position;
+import org.apache.nifi.flow.ScheduledState;
import org.apache.nifi.flow.VersionedConnection;
+import org.apache.nifi.flow.VersionedControllerService;
import org.apache.nifi.flow.VersionedLabel;
import org.apache.nifi.flow.VersionedProcessGroup;
import org.apache.nifi.flow.VersionedProcessor;
@@ -43,6 +46,27 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
class RebaseEngineTest {
+ private static final String ROOT_ID = "root";
+ private static final String VERSION_N_ROOT_ID = "root-n";
+ private static final String TARGET_ROOT_ID = "root-n-plus-one";
+ private static final String PROCESSOR_ID = "proc-a";
+ private static final String PROCESSOR_NAME = "ProcessorA";
+ private static final String SERVICE_A_ID = "service-a";
+ private static final String SERVICE_X_ID = "service-x";
+ private static final String SERVICE_Y_ID = "service-y";
+ private static final String SERVICE_Z_ID = "service-z";
+ private static final String SERVICE_A_NAME = "Service A";
+ private static final String SERVICE_X_NAME = "Service X";
+ private static final String SERVICE_Y_NAME = "Service Y";
+ private static final String SERVICE_Z_NAME = "Service Z";
+ private static final String CONTROLLER_SERVICE_PROPERTY =
"controller.service";
+ private static final String DYNAMIC_Y_PROPERTY = "dynamic.y";
+ private static final String DYNAMIC_Z_PROPERTY = "dynamic.z";
+ private static final String CONTROLLER_SERVICE_TYPE =
"org.apache.nifi.services.LocalControllerService";
+ private static final String BUNDLE_GROUP = "group";
+ private static final String BUNDLE_ARTIFACT = "artifact";
+ private static final String BUNDLE_VERSION = "1.0.0";
+
private RebaseEngine engine;
@BeforeEach
@@ -142,7 +166,7 @@ class RebaseEngineTest {
}
@Test
- void testUnsupportedDifferenceTypeNoHandler() {
+ void
testUnsupportedLocalProcessorAdditionUsesRegisteredComponentAddedHandler() {
final VersionedProcessor processor = createProcessor("proc-a",
"ProcessorA");
final Set<FlowDifference> localDifferences = new HashSet<>();
@@ -159,8 +183,164 @@ class RebaseEngineTest {
final RebaseAnalysis.ClassifiedDifference classified =
analysis.getClassifiedLocalChanges().get(0);
assertEquals(RebaseClassification.UNSUPPORTED,
classified.getClassification());
- assertEquals(RebaseConflictCode.NO_HANDLER,
classified.getConflictCode());
+ assertEquals(RebaseConflictCode.UNSUPPORTED_COMPONENT_TYPE,
classified.getConflictCode());
+ assertNull(analysis.getMergedSnapshot());
+ }
+
+ @Test
+ void
testAnalyzeScenario1PreservesAddedControllerServiceAndProcessorReference() {
+ final VersionedControllerService versionNService =
createControllerService(SERVICE_X_ID, SERVICE_X_NAME, ROOT_ID);
+ final VersionedControllerService localAddedService =
createControllerService(SERVICE_Y_ID, SERVICE_Y_NAME, ROOT_ID);
+
+ final VersionedProcessor versionNProcessor =
createProcessorWithProperty(PROCESSOR_ID, PROCESSOR_NAME,
CONTROLLER_SERVICE_PROPERTY, SERVICE_X_ID);
+ final VersionedProcessor localProcessor =
createProcessorWithProperty(PROCESSOR_ID, PROCESSOR_NAME,
CONTROLLER_SERVICE_PROPERTY, SERVICE_Y_ID);
+
+ final Set<FlowDifference> localDifferences = new HashSet<>();
+ localDifferences.add(new
StandardFlowDifference(DifferenceType.COMPONENT_ADDED, null, localAddedService,
+ null, localAddedService, "Controller service added locally"));
+ localDifferences.add(new
StandardFlowDifference(DifferenceType.PROPERTY_CHANGED, versionNProcessor,
localProcessor, CONTROLLER_SERVICE_PROPERTY,
+ SERVICE_X_ID, SERVICE_Y_ID, "Processor property changed
locally"));
+
+ final VersionedProcessGroup targetSnapshot = new
VersionedProcessGroup();
+ targetSnapshot.setIdentifier(ROOT_ID);
+ targetSnapshot.getControllerServices().add(versionNService);
+
targetSnapshot.getProcessors().add(createProcessorWithProperty(PROCESSOR_ID,
PROCESSOR_NAME, CONTROLLER_SERVICE_PROPERTY, SERVICE_X_ID));
+
+ final RebaseAnalysis analysis = engine.analyze(localDifferences,
Collections.emptySet(), targetSnapshot);
+
+ assertTrue(analysis.isRebaseAllowed());
+ assertEquals(2, analysis.getClassifiedLocalChanges().size());
+ assertAllCompatible(analysis);
+ assertClassification(analysis, DifferenceType.COMPONENT_ADDED,
SERVICE_Y_ID, RebaseClassification.COMPATIBLE, null);
+ assertClassification(analysis, DifferenceType.PROPERTY_CHANGED,
PROCESSOR_ID, RebaseClassification.COMPATIBLE, null);
+
+ final VersionedProcessGroup merged = analysis.getMergedSnapshot();
+ assertNotNull(merged);
+ final VersionedControllerService mergedService =
findControllerServiceById(merged, SERVICE_Y_ID);
+ assertSame(localAddedService, mergedService);
+
+ final VersionedProcessor mergedProcessor = findProcessorById(merged,
PROCESSOR_ID);
+ assertNotNull(mergedProcessor);
+ assertEquals(SERVICE_Y_ID,
mergedProcessor.getProperties().get(CONTROLLER_SERVICE_PROPERTY));
+ }
+
+ @Test
+ void
testAnalyzePreservesRootAdditionWhenTargetRootIdentifierChangedUpstream() {
+ final VersionedControllerService localAddedService =
createControllerService(SERVICE_Y_ID, SERVICE_Y_NAME, VERSION_N_ROOT_ID);
+ final Set<FlowDifference> localDifferences = Set.of(new
StandardFlowDifference(DifferenceType.COMPONENT_ADDED, null, localAddedService,
+ null, localAddedService, "Controller service added locally"));
+
+ final VersionedProcessGroup targetRoot = new VersionedProcessGroup();
+ targetRoot.setIdentifier(TARGET_ROOT_ID);
+ targetRoot.setName("Root");
+
+ final RebaseAnalysis analysis = engine.analyze(localDifferences,
Collections.emptySet(), targetRoot);
+
+ assertTrue(analysis.isRebaseAllowed());
+ assertSame(localAddedService, findControllerServiceById(targetRoot,
SERVICE_Y_ID));
+ assertEquals(TARGET_ROOT_ID, localAddedService.getGroupIdentifier());
+ }
+
+ @Test
+ void testAnalyzeRejectsRootAdditionWhenParentRemovedUpstream() {
+ final VersionedControllerService localAddedService =
createControllerService(SERVICE_Y_ID, SERVICE_Y_NAME, VERSION_N_ROOT_ID);
+ final Set<FlowDifference> localDifferences = Set.of(new
StandardFlowDifference(DifferenceType.COMPONENT_ADDED, null, localAddedService,
+ null, localAddedService, "Controller service added locally"));
+
+ final VersionedProcessGroup removedParent = new
VersionedProcessGroup();
+ removedParent.setIdentifier(VERSION_N_ROOT_ID);
+
+ final VersionedProcessGroup targetRoot = new VersionedProcessGroup();
+ targetRoot.setIdentifier("replacement-root");
+
+ final Set<FlowDifference> upstreamDifferences = Set.of(new
StandardFlowDifference(DifferenceType.COMPONENT_REMOVED, removedParent, null,
+ removedParent, null, "Parent removed upstream"));
+
+ final RebaseAnalysis analysis = engine.analyze(localDifferences,
upstreamDifferences, targetRoot);
+
+ assertFalse(analysis.isRebaseAllowed());
+ assertClassification(analysis, DifferenceType.COMPONENT_ADDED,
SERVICE_Y_ID, RebaseClassification.UNSUPPORTED,
+ RebaseConflictCode.COMPONENT_NOT_FOUND);
+ }
+
+ @Test
+ void
testAnalyzeScenario2PreservesMultipleAddedControllerServicesAndDynamicReferences()
{
+ final VersionedControllerService versionNServiceA =
createControllerService(SERVICE_A_ID, SERVICE_A_NAME, ROOT_ID);
+ versionNServiceA.setProperties(Collections.emptyMap());
+ versionNServiceA.setPropertyDescriptors(Collections.emptyMap());
+
+ final VersionedControllerService localServiceA =
createControllerService(SERVICE_A_ID, SERVICE_A_NAME, ROOT_ID);
+ localServiceA.setProperties(Map.of(DYNAMIC_Y_PROPERTY, SERVICE_Y_ID,
DYNAMIC_Z_PROPERTY, SERVICE_Z_ID));
+ localServiceA.setPropertyDescriptors(Map.of(
+ DYNAMIC_Y_PROPERTY,
createPropertyDescriptor(DYNAMIC_Y_PROPERTY, true, false),
+ DYNAMIC_Z_PROPERTY,
createPropertyDescriptor(DYNAMIC_Z_PROPERTY, true, false)));
+
+ final VersionedControllerService localServiceY =
createControllerService(SERVICE_Y_ID, SERVICE_Y_NAME, ROOT_ID);
+ final VersionedControllerService localServiceZ =
createControllerService(SERVICE_Z_ID, SERVICE_Z_NAME, ROOT_ID);
+
+ final Set<FlowDifference> localDifferences = new HashSet<>();
+ localDifferences.add(new
StandardFlowDifference(DifferenceType.COMPONENT_ADDED, null, localServiceY,
+ null, localServiceY, "Controller service Y added locally"));
+ localDifferences.add(new
StandardFlowDifference(DifferenceType.COMPONENT_ADDED, null, localServiceZ,
+ null, localServiceZ, "Controller service Z added locally"));
+ localDifferences.add(new
StandardFlowDifference(DifferenceType.PROPERTY_ADDED, versionNServiceA,
localServiceA, DYNAMIC_Y_PROPERTY,
+ null, SERVICE_Y_ID, "Dynamic property added for service Y"));
+ localDifferences.add(new
StandardFlowDifference(DifferenceType.PROPERTY_ADDED, versionNServiceA,
localServiceA, DYNAMIC_Z_PROPERTY,
+ null, SERVICE_Z_ID, "Dynamic property added for service Z"));
+
+ final VersionedProcessGroup targetSnapshot = new
VersionedProcessGroup();
+ targetSnapshot.setIdentifier(ROOT_ID);
+ targetSnapshot.getControllerServices().add(versionNServiceA);
+
+ final RebaseAnalysis analysis = engine.analyze(localDifferences,
Collections.emptySet(), targetSnapshot);
+
+ assertTrue(analysis.isRebaseAllowed());
+ assertEquals(4, analysis.getClassifiedLocalChanges().size());
+ assertAllCompatible(analysis);
+ assertClassification(analysis, DifferenceType.COMPONENT_ADDED,
SERVICE_Y_ID, RebaseClassification.COMPATIBLE, null);
+ assertClassification(analysis, DifferenceType.COMPONENT_ADDED,
SERVICE_Z_ID, RebaseClassification.COMPATIBLE, null);
+
+ final VersionedProcessGroup merged = analysis.getMergedSnapshot();
+ assertNotNull(merged);
+ assertSame(localServiceY, findControllerServiceById(merged,
SERVICE_Y_ID));
+ assertSame(localServiceZ, findControllerServiceById(merged,
SERVICE_Z_ID));
+
+ final VersionedControllerService mergedServiceA =
findControllerServiceById(merged, SERVICE_A_ID);
+ assertNotNull(mergedServiceA);
+ assertEquals(SERVICE_Y_ID,
mergedServiceA.getProperties().get(DYNAMIC_Y_PROPERTY));
+ assertEquals(SERVICE_Z_ID,
mergedServiceA.getProperties().get(DYNAMIC_Z_PROPERTY));
+ }
+
+ @Test
+ void testAnalyzeCollisionBlocksRebaseAndDoesNotMutateTargetSnapshot() {
+ final VersionedControllerService collidingTargetService =
createControllerService(SERVICE_Y_ID, "Target Service Y", ROOT_ID);
+ final VersionedControllerService localAddedService =
createControllerService(SERVICE_Y_ID, "Local Service Y", ROOT_ID);
+
+ final VersionedProcessor versionNProcessor =
createProcessorWithProperty(PROCESSOR_ID, PROCESSOR_NAME,
CONTROLLER_SERVICE_PROPERTY, SERVICE_X_ID);
+ final VersionedProcessor localProcessor =
createProcessorWithProperty(PROCESSOR_ID, PROCESSOR_NAME,
CONTROLLER_SERVICE_PROPERTY, SERVICE_Y_ID);
+
+ final Set<FlowDifference> localDifferences = new HashSet<>();
+ localDifferences.add(new
StandardFlowDifference(DifferenceType.COMPONENT_ADDED, null, localAddedService,
+ null, localAddedService, "Controller service added with
collision"));
+ localDifferences.add(new
StandardFlowDifference(DifferenceType.PROPERTY_CHANGED, versionNProcessor,
localProcessor, CONTROLLER_SERVICE_PROPERTY,
+ SERVICE_X_ID, SERVICE_Y_ID, "Processor property changed
locally"));
+
+ final VersionedProcessGroup targetSnapshot = new
VersionedProcessGroup();
+ targetSnapshot.setIdentifier(ROOT_ID);
+ targetSnapshot.getControllerServices().add(collidingTargetService);
+
targetSnapshot.getProcessors().add(createProcessorWithProperty(PROCESSOR_ID,
PROCESSOR_NAME, CONTROLLER_SERVICE_PROPERTY, SERVICE_X_ID));
+
+ final RebaseAnalysis analysis = engine.analyze(localDifferences,
Collections.emptySet(), targetSnapshot);
+
+ assertFalse(analysis.isRebaseAllowed());
assertNull(analysis.getMergedSnapshot());
+ assertClassification(analysis, DifferenceType.COMPONENT_ADDED,
SERVICE_Y_ID, RebaseClassification.CONFLICTING,
+ RebaseConflictCode.COMPONENT_IDENTIFIER_COLLISION);
+ assertEquals(1, countComponentsById(targetSnapshot, SERVICE_Y_ID));
+
+ final VersionedProcessor unchangedProcessor =
findProcessorById(targetSnapshot, PROCESSOR_ID);
+ assertNotNull(unchangedProcessor);
+ assertEquals(SERVICE_X_ID,
unchangedProcessor.getProperties().get(CONTROLLER_SERVICE_PROPERTY));
}
@Test
@@ -652,6 +832,28 @@ class RebaseEngineTest {
return connection;
}
+ private VersionedControllerService createControllerService(final String
identifier, final String name, final String groupIdentifier) {
+ final VersionedControllerService service = new
VersionedControllerService();
+ service.setIdentifier(identifier);
+ service.setName(name);
+ service.setGroupIdentifier(groupIdentifier);
+ service.setType(CONTROLLER_SERVICE_TYPE);
+ service.setBundle(new Bundle(BUNDLE_GROUP, BUNDLE_ARTIFACT,
BUNDLE_VERSION));
+ service.setScheduledState(ScheduledState.DISABLED);
+ service.setComments(name + " comments");
+ service.setProperties(Collections.emptyMap());
+ service.setPropertyDescriptors(Collections.emptyMap());
+ return service;
+ }
+
+ private VersionedPropertyDescriptor createPropertyDescriptor(final String
propertyName, final boolean dynamic, final boolean sensitive) {
+ final VersionedPropertyDescriptor descriptor = new
VersionedPropertyDescriptor();
+ descriptor.setName(propertyName);
+ descriptor.setDynamic(dynamic);
+ descriptor.setSensitive(sensitive);
+ return descriptor;
+ }
+
private VersionedProcessor findProcessorById(final VersionedProcessGroup
group, final String identifier) {
for (final VersionedProcessor processor : group.getProcessors()) {
if (identifier.equals(processor.getIdentifier())) {
@@ -666,4 +868,64 @@ class RebaseEngineTest {
}
return null;
}
+
+ private VersionedControllerService findControllerServiceById(final
VersionedProcessGroup group, final String identifier) {
+ for (final VersionedControllerService service :
group.getControllerServices()) {
+ if (identifier.equals(service.getIdentifier())) {
+ return service;
+ }
+ }
+ for (final VersionedProcessGroup childGroup :
group.getProcessGroups()) {
+ final VersionedControllerService result =
findControllerServiceById(childGroup, identifier);
+ if (result != null) {
+ return result;
+ }
+ }
+ return null;
+ }
+
+ private int countComponentsById(final VersionedProcessGroup group, final
String identifier) {
+ int count = identifier.equals(group.getIdentifier()) ? 1 : 0;
+
+ for (final VersionedProcessor processor : group.getProcessors()) {
+ if (identifier.equals(processor.getIdentifier())) {
+ count++;
+ }
+ }
+ for (final VersionedControllerService service :
group.getControllerServices()) {
+ if (identifier.equals(service.getIdentifier())) {
+ count++;
+ }
+ }
+ for (final VersionedProcessGroup childGroup :
group.getProcessGroups()) {
+ count += countComponentsById(childGroup, identifier);
+ }
+
+ return count;
+ }
+
+ private void assertAllCompatible(final RebaseAnalysis analysis) {
+ for (final RebaseAnalysis.ClassifiedDifference classified :
analysis.getClassifiedLocalChanges()) {
+ assertEquals(RebaseClassification.COMPATIBLE,
classified.getClassification());
+ }
+ }
+
+ private void assertClassification(final RebaseAnalysis analysis, final
DifferenceType differenceType, final String componentIdentifier,
+ final RebaseClassification
expectedClassification, final RebaseConflictCode expectedConflictCode) {
+ RebaseAnalysis.ClassifiedDifference matchingDifference = null;
+ for (final RebaseAnalysis.ClassifiedDifference classified :
analysis.getClassifiedLocalChanges()) {
+ final FlowDifference difference = classified.getDifference();
+ final String differenceComponentId = difference.getComponentB() !=
null
+ ? difference.getComponentB().getIdentifier()
+ : difference.getComponentA().getIdentifier();
+ if (difference.getDifferenceType() == differenceType &&
componentIdentifier.equals(differenceComponentId)) {
+ matchingDifference = classified;
+ break;
+ }
+ }
+
+ assertNotNull(matchingDifference);
+ assertEquals(expectedClassification,
matchingDifference.getClassification());
+ assertEquals(expectedConflictCode,
matchingDifference.getConflictCode());
+ }
}
diff --git
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions-services/src/main/java/org/apache/nifi/cs/tests/system/FakeDynamicPropertiesControllerService.java
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions-services/src/main/java/org/apache/nifi/cs/tests/system/FakeDynamicPropertiesControllerService.java
new file mode 100644
index 00000000000..000675731c5
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions-services/src/main/java/org/apache/nifi/cs/tests/system/FakeDynamicPropertiesControllerService.java
@@ -0,0 +1,46 @@
+/*
+ * 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.cs.tests.system;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.controller.AbstractControllerService;
+
+public class FakeDynamicPropertiesControllerService extends
AbstractControllerService implements BaseFakeService {
+ @Override
+ protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final
String propertyName) {
+ if (propertyName.startsWith("FCS.")) {
+ return new PropertyDescriptor.Builder()
+ .name(propertyName)
+ .required(false)
+ .dynamic(true)
+ .identifiesControllerService(BaseFakeService.class)
+ .build();
+ }
+
+ return new PropertyDescriptor.Builder()
+ .name(propertyName)
+ .required(false)
+ .addValidator(Validator.VALID)
+ .dynamic(true)
+ .build();
+ }
+
+ @Override
+ public void foo() {
+ }
+}
diff --git
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions-services/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions-services/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
index 3e376b35485..b4bda8321c0 100644
---
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions-services/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
+++
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions-services/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
@@ -15,6 +15,7 @@
org.apache.nifi.cs.tests.system.ClassloaderIsolationKeyProviderService
org.apache.nifi.cs.tests.system.EnsureControllerServiceConfigurationCorrect
+org.apache.nifi.cs.tests.system.FakeDynamicPropertiesControllerService
org.apache.nifi.cs.tests.system.FakeControllerService1
org.apache.nifi.cs.tests.system.LifecycleFailureService
org.apache.nifi.cs.tests.system.SensitiveDynamicPropertiesService
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/registry/RebaseVersionIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/registry/RebaseVersionIT.java
index c698dae31bb..ae57b15dbe6 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/registry/RebaseVersionIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/registry/RebaseVersionIT.java
@@ -20,9 +20,12 @@ package org.apache.nifi.tests.system.registry;
import org.apache.nifi.tests.system.NiFiClientUtil;
import org.apache.nifi.tests.system.NiFiSystemIT;
import org.apache.nifi.toolkit.client.NiFiClientException;
+import org.apache.nifi.web.api.dto.ComponentDifferenceDTO;
+import org.apache.nifi.web.api.dto.DifferenceDTO;
import org.apache.nifi.web.api.dto.RebaseChangeDTO;
import org.apache.nifi.web.api.dto.VersionControlInformationDTO;
import org.apache.nifi.web.api.dto.flow.FlowDTO;
+import org.apache.nifi.web.api.entity.ControllerServiceEntity;
import org.apache.nifi.web.api.entity.FlowComparisonEntity;
import org.apache.nifi.web.api.entity.FlowRegistryClientEntity;
import org.apache.nifi.web.api.entity.ProcessGroupEntity;
@@ -34,6 +37,8 @@ import
org.apache.nifi.web.api.entity.VersionedFlowUpdateRequestEntity;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
import java.util.Map;
import java.util.Set;
@@ -44,6 +49,19 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class RebaseVersionIT extends NiFiSystemIT {
private static final String TEST_FLOWS_BUCKET = "test-flows";
+ private static final String ROOT_GROUP_ID = "root";
+ private static final String ORIGINAL_GROUP_NAME = "Original";
+ private static final String CONTROLLER_SERVICE_TYPE =
"FakeControllerService1";
+ private static final String DYNAMIC_CONTROLLER_SERVICE_TYPE =
"FakeDynamicPropertiesControllerService";
+ private static final String PROCESSOR_TYPE = "FakeProcessor";
+ private static final String GENERATE_FLOW_FILE_TYPE = "GenerateFlowFile";
+ private static final String CONTROLLER_SERVICE_PROPERTY = "Fake Service";
+ private static final String TEXT_PROPERTY = "Text";
+ private static final String UPSTREAM_CHANGE = "upstream-change";
+ private static final String SERVICE_X_PROPERTY = "FCS.X";
+ private static final String SERVICE_Y_PROPERTY = "FCS.Y";
+ private static final String SERVICE_Z_PROPERTY = "FCS.Z";
+ private static final String TARGET_VERSION = "2";
@Test
public void testCleanRebaseWithPositionAndPropertyChanges() throws
NiFiClientException, IOException, InterruptedException {
@@ -63,14 +81,14 @@ public class RebaseVersionIT extends NiFiSystemIT {
util.updateProcessorProperties(generate, Map.of("File Size", "99 B"));
- final RebaseAnalysisEntity analysis =
util.getRebaseAnalysis(originalGroup.getId(), "2");
+ final RebaseAnalysisEntity analysis =
util.getRebaseAnalysis(originalGroup.getId(), TARGET_VERSION);
assertTrue(analysis.getRebaseAllowed(), "Expected rebase to be allowed
but it was not. Failure: " + analysis.getFailureReason()
+ ". Local changes: " + describeLocalChanges(analysis));
- util.rebaseFlowVersion(originalGroup.getId(), "2");
+ util.rebaseFlowVersion(originalGroup.getId(), TARGET_VERSION);
final VersionControlInformationDTO updatedVci =
getVersionControlInfo(originalGroup.getId());
- assertEquals("2", updatedVci.getVersion());
+ assertEquals(TARGET_VERSION, updatedVci.getVersion());
final ProcessorEntity rebasedProcessor =
findSingleProcessor(originalGroup.getId());
final Map<String, String> properties =
rebasedProcessor.getComponent().getConfig().getProperties();
@@ -96,7 +114,7 @@ public class RebaseVersionIT extends NiFiSystemIT {
util.updateProcessorProperties(generate, Map.of("File Size", "50 B"));
- final RebaseAnalysisEntity analysis =
util.getRebaseAnalysis(originalGroup.getId(), "2");
+ final RebaseAnalysisEntity analysis =
util.getRebaseAnalysis(originalGroup.getId(), TARGET_VERSION);
assertFalse(analysis.getRebaseAllowed());
final boolean hasConflicting = analysis.getLocalChanges().stream()
@@ -123,7 +141,7 @@ public class RebaseVersionIT extends NiFiSystemIT {
util.createProcessor("TerminateFlowFile", originalGroup.getId());
- final RebaseAnalysisEntity analysis =
util.getRebaseAnalysis(originalGroup.getId(), "2");
+ final RebaseAnalysisEntity analysis =
util.getRebaseAnalysis(originalGroup.getId(), TARGET_VERSION);
assertFalse(analysis.getRebaseAllowed());
final boolean hasUnsupported = analysis.getLocalChanges().stream()
@@ -150,10 +168,10 @@ public class RebaseVersionIT extends NiFiSystemIT {
util.updateProcessorProperties(generate, Map.of("File Size", "99 B"));
- util.rebaseFlowVersion(originalGroup.getId(), "2");
+ util.rebaseFlowVersion(originalGroup.getId(), TARGET_VERSION);
final VersionControlInformationDTO afterRebase =
getVersionControlInfo(originalGroup.getId());
- assertEquals("2", afterRebase.getVersion());
+ assertEquals(TARGET_VERSION, afterRebase.getVersion());
util.saveFlowVersion(originalGroup, clientEntity,
getVersionControlInformation(originalGroup.getId()));
@@ -179,15 +197,15 @@ public class RebaseVersionIT extends NiFiSystemIT {
util.updateProcessorProperties(generate, Map.of("File Size", "99 B"));
- util.rebaseFlowVersion(originalGroup.getId(), "2");
+ util.rebaseFlowVersion(originalGroup.getId(), TARGET_VERSION);
final VersionControlInformationDTO afterRebase =
getVersionControlInfo(originalGroup.getId());
- assertEquals("2", afterRebase.getVersion());
+ assertEquals(TARGET_VERSION, afterRebase.getVersion());
util.revertChanges(originalGroup);
final VersionControlInformationDTO revertedVci =
getVersionControlInfo(originalGroup.getId());
- assertEquals("2", revertedVci.getVersion());
+ assertEquals(TARGET_VERSION, revertedVci.getVersion());
}
@Test
@@ -252,23 +270,23 @@ public class RebaseVersionIT extends NiFiSystemIT {
util.updateProcessorProperties(generate, Map.of("File Size", "99 B"));
- final RebaseAnalysisEntity initialAnalysis =
util.getRebaseAnalysis(originalGroup.getId(), "2");
+ final RebaseAnalysisEntity initialAnalysis =
util.getRebaseAnalysis(originalGroup.getId(), TARGET_VERSION);
final String staleFingerprint =
initialAnalysis.getAnalysisFingerprint();
util.updateProcessorProperties(generate, Map.of("Max FlowFiles",
"50"));
boolean rebaseWithStaleFailed = false;
try {
- executeRebaseWithFingerprint(originalGroup, "2", staleFingerprint);
+ executeRebaseWithFingerprint(originalGroup, TARGET_VERSION,
staleFingerprint);
} catch (final Exception e) {
rebaseWithStaleFailed = true;
}
assertTrue(rebaseWithStaleFailed);
- util.rebaseFlowVersion(originalGroup.getId(), "2");
+ util.rebaseFlowVersion(originalGroup.getId(), TARGET_VERSION);
final VersionControlInformationDTO updatedVci =
getVersionControlInfo(originalGroup.getId());
- assertEquals("2", updatedVci.getVersion());
+ assertEquals(TARGET_VERSION, updatedVci.getVersion());
}
@Test
@@ -293,7 +311,7 @@ public class RebaseVersionIT extends NiFiSystemIT {
util.updateProcessorProperties(secondParentProcessor, Map.of("File
Size", "20 B"));
util.saveFlowVersion(secondParent, clientEntity,
getVersionControlInformation(secondParent.getId()));
- final RebaseAnalysisEntity analysis =
util.getRebaseAnalysis(parentGroup.getId(), "2");
+ final RebaseAnalysisEntity analysis =
util.getRebaseAnalysis(parentGroup.getId(), TARGET_VERSION);
assertFalse(analysis.getRebaseAllowed(), "Rebase should be blocked due
to descendant modifications");
assertNotNull(analysis.getFailureReason());
}
@@ -315,10 +333,10 @@ public class RebaseVersionIT extends NiFiSystemIT {
util.saveFlowVersion(secondGroup, clientEntity,
getVersionControlInformation(secondGroup.getId()));
util.updateProcessorProperties(generate, Map.of("File Size", "99 B"));
- util.rebaseFlowVersion(originalGroup.getId(), "2");
+ util.rebaseFlowVersion(originalGroup.getId(), TARGET_VERSION);
final VersionControlInformationDTO afterFirstRebase =
getVersionControlInfo(originalGroup.getId());
- assertEquals("2", afterFirstRebase.getVersion());
+ assertEquals(TARGET_VERSION, afterFirstRebase.getVersion());
final ProcessorEntity v3Processor =
findSingleProcessor(secondGroup.getId());
util.updateProcessorProperties(v3Processor, Map.of("Max FlowFiles",
"30"));
@@ -356,7 +374,7 @@ public class RebaseVersionIT extends NiFiSystemIT {
// Locally modify the processor that the target version removed
util.updateProcessorProperties(generate, Map.of("File Size", "99 B"));
- final RebaseAnalysisEntity analysis =
util.getRebaseAnalysis(originalGroup.getId(), "2");
+ final RebaseAnalysisEntity analysis =
util.getRebaseAnalysis(originalGroup.getId(), TARGET_VERSION);
assertFalse(analysis.getRebaseAllowed(), "Rebase should be blocked
because the target version removed the locally modified component");
final boolean removedComponentRejected =
analysis.getLocalChanges().stream()
@@ -383,10 +401,10 @@ public class RebaseVersionIT extends NiFiSystemIT {
util.updateProcessorProperties(generate, Map.of("File Size", "99 B"));
- util.rebaseFlowVersion(originalGroup.getId(), "2");
+ util.rebaseFlowVersion(originalGroup.getId(), TARGET_VERSION);
final VersionControlInformationDTO updatedVci =
getVersionControlInfo(originalGroup.getId());
- assertEquals("2", updatedVci.getVersion());
+ assertEquals(TARGET_VERSION, updatedVci.getVersion());
// The Version Control Information snapshot must be the clean target
version (not the merged snapshot). If it were
// the merged snapshot, the preserved local change would not be
reported as a local modification. This is the key
@@ -396,6 +414,98 @@ public class RebaseVersionIT extends NiFiSystemIT {
"Expected the preserved local change to be reported as a local
modification after rebase, but none were found");
}
+ @Test
+ public void
testRebasePreservesLocallyAddedControllerServiceReferencedByProcessor() throws
NiFiClientException, IOException, InterruptedException {
+ final FlowRegistryClientEntity clientEntity = registerClient();
+ final NiFiClientUtil util = getClientUtil();
+
+ final ProcessGroupEntity originalGroup =
util.createProcessGroup(ORIGINAL_GROUP_NAME, ROOT_GROUP_ID);
+ final ControllerServiceEntity serviceX =
util.createControllerService(CONTROLLER_SERVICE_TYPE, originalGroup.getId());
+ final ProcessorEntity fakeProcessor =
util.createProcessor(PROCESSOR_TYPE, originalGroup.getId());
+ util.updateProcessorProperties(fakeProcessor,
Map.of(CONTROLLER_SERVICE_PROPERTY, serviceX.getId()));
+ util.createProcessor(GENERATE_FLOW_FILE_TYPE, originalGroup.getId());
+
+ final VersionControlInformationEntity vci =
util.startVersionControl(originalGroup, clientEntity, TEST_FLOWS_BUCKET,
+ "RebaseLocalAddedControllerServiceProcessorReference");
+ final String flowId = vci.getVersionControlInformation().getFlowId();
+
+ final ProcessGroupEntity secondGroup =
util.importFlowFromRegistry(ROOT_GROUP_ID, clientEntity.getId(),
TEST_FLOWS_BUCKET, flowId, "1");
+ final ProcessorEntity upstreamGenerate =
findProcessorByType(secondGroup.getId(), GENERATE_FLOW_FILE_TYPE);
+ util.updateProcessorProperties(upstreamGenerate, Map.of(TEXT_PROPERTY,
UPSTREAM_CHANGE));
+ util.saveFlowVersion(secondGroup, clientEntity,
getVersionControlInformation(secondGroup.getId()));
+
+ final ControllerServiceEntity serviceY =
util.createControllerService(CONTROLLER_SERVICE_TYPE, originalGroup.getId());
+ util.updateProcessorProperties(fakeProcessor,
Map.of(CONTROLLER_SERVICE_PROPERTY, serviceY.getId()));
+
+ final RebaseAnalysisEntity analysis =
util.getRebaseAnalysis(originalGroup.getId(), TARGET_VERSION);
+ assertTrue(analysis.getRebaseAllowed(), "Expected rebase to be allowed
but it was not. Failure: " + analysis.getFailureReason()
+ + ". Local changes: " + describeLocalChanges(analysis));
+ assertCompatibleControllerServiceAdditions(analysis, 1);
+
+ util.rebaseFlowVersion(originalGroup.getId(), TARGET_VERSION);
+
+ final VersionControlInformationDTO updatedVci =
getVersionControlInfo(originalGroup.getId());
+ assertEquals(TARGET_VERSION, updatedVci.getVersion());
+
+ final Set<ControllerServiceEntity> rebasedServices =
getNifiClient().getFlowClient().getControllerServices(originalGroup.getId()).getControllerServices();
+ assertControllerServicesPresent(rebasedServices, serviceX.getId(),
serviceY.getId());
+
+ final ProcessorEntity rebasedProcessor =
findProcessorByType(originalGroup.getId(), PROCESSOR_TYPE);
+ assertEquals(serviceY.getId(),
rebasedProcessor.getComponent().getConfig().getProperties().get(CONTROLLER_SERVICE_PROPERTY));
+
+ assertLocalModificationsContainComponents(originalGroup.getId(),
serviceY.getId(), fakeProcessor.getId());
+ }
+
+ @Test
+ public void
testRebasePreservesLocallyAddedControllerServicesReferencedByDynamicControllerServiceProperties()
+ throws NiFiClientException, IOException, InterruptedException {
+ final FlowRegistryClientEntity clientEntity = registerClient();
+ final NiFiClientUtil util = getClientUtil();
+
+ final ProcessGroupEntity originalGroup =
util.createProcessGroup(ORIGINAL_GROUP_NAME, ROOT_GROUP_ID);
+ final ControllerServiceEntity serviceX =
util.createControllerService(CONTROLLER_SERVICE_TYPE, originalGroup.getId());
+ final ControllerServiceEntity dynamicService =
util.createControllerService(DYNAMIC_CONTROLLER_SERVICE_TYPE,
originalGroup.getId());
+ util.updateControllerServiceProperties(dynamicService,
Collections.singletonMap(SERVICE_X_PROPERTY, serviceX.getId()));
+ util.createProcessor(GENERATE_FLOW_FILE_TYPE, originalGroup.getId());
+
+ final VersionControlInformationEntity vci =
util.startVersionControl(originalGroup, clientEntity, TEST_FLOWS_BUCKET,
+ "RebaseLocalAddedControllerServiceDynamicReference");
+ final String flowId = vci.getVersionControlInformation().getFlowId();
+
+ final ProcessGroupEntity secondGroup =
util.importFlowFromRegistry(ROOT_GROUP_ID, clientEntity.getId(),
TEST_FLOWS_BUCKET, flowId, "1");
+ final ProcessorEntity upstreamGenerate =
findProcessorByType(secondGroup.getId(), GENERATE_FLOW_FILE_TYPE);
+ util.updateProcessorProperties(upstreamGenerate, Map.of(TEXT_PROPERTY,
UPSTREAM_CHANGE));
+ util.saveFlowVersion(secondGroup, clientEntity,
getVersionControlInformation(secondGroup.getId()));
+
+ final ControllerServiceEntity serviceY =
util.createControllerService(CONTROLLER_SERVICE_TYPE, originalGroup.getId());
+ final ControllerServiceEntity serviceZ =
util.createControllerService(CONTROLLER_SERVICE_TYPE, originalGroup.getId());
+ util.updateControllerServiceProperties(dynamicService, Map.of(
+ SERVICE_X_PROPERTY, serviceX.getId(),
+ SERVICE_Y_PROPERTY, serviceY.getId(),
+ SERVICE_Z_PROPERTY, serviceZ.getId()));
+
+ final RebaseAnalysisEntity analysis =
util.getRebaseAnalysis(originalGroup.getId(), TARGET_VERSION);
+ assertTrue(analysis.getRebaseAllowed(), "Expected rebase to be allowed
but it was not. Failure: " + analysis.getFailureReason()
+ + ". Local changes: " + describeLocalChanges(analysis));
+ assertCompatibleControllerServiceAdditions(analysis, 2);
+
+ util.rebaseFlowVersion(originalGroup.getId(), TARGET_VERSION);
+
+ final VersionControlInformationDTO updatedVci =
getVersionControlInfo(originalGroup.getId());
+ assertEquals(TARGET_VERSION, updatedVci.getVersion());
+
+ final Set<ControllerServiceEntity> rebasedServices =
getNifiClient().getFlowClient().getControllerServices(originalGroup.getId()).getControllerServices();
+ assertControllerServicesPresent(rebasedServices,
dynamicService.getId(), serviceX.getId(), serviceY.getId(), serviceZ.getId());
+
+ final ControllerServiceEntity rebasedDynamicService =
getNifiClient().getControllerServicesClient().getControllerService(dynamicService.getId());
+ final Map<String, String> dynamicProperties =
rebasedDynamicService.getComponent().getProperties();
+ assertEquals(serviceX.getId(),
dynamicProperties.get(SERVICE_X_PROPERTY));
+ assertEquals(serviceY.getId(),
dynamicProperties.get(SERVICE_Y_PROPERTY));
+ assertEquals(serviceZ.getId(),
dynamicProperties.get(SERVICE_Z_PROPERTY));
+
+ assertLocalModificationsContainComponents(originalGroup.getId(),
dynamicService.getId(), serviceY.getId(), serviceZ.getId());
+ }
+
private String describeLocalChanges(final RebaseAnalysisEntity analysis) {
if (analysis.getLocalChanges() == null ||
analysis.getLocalChanges().isEmpty()) {
return "none";
@@ -434,6 +544,74 @@ public class RebaseVersionIT extends NiFiSystemIT {
.orElseThrow(() -> new AssertionError("No processor of type "
+ simpleTypeName + " found in group " + processGroupId));
}
+ private void assertCompatibleControllerServiceAdditions(final
RebaseAnalysisEntity analysis, final long expectedCount) {
+ final long compatibleAdditions = analysis.getLocalChanges().stream()
+ .filter(change -> "Component
Added".equals(change.getDifferenceType()))
+ .filter(change -> "Controller
Service".equals(change.getComponentType()))
+ .filter(change ->
"COMPATIBLE".equals(change.getClassification()))
+ .count();
+ assertEquals(expectedCount, compatibleAdditions, "Unexpected
compatible controller-service additions. Local changes: "
+ + describeLocalChanges(analysis));
+ }
+
+ private void assertControllerServicesPresent(final
Set<ControllerServiceEntity> services, final String... expectedServiceIds) {
+ for (final String serviceId : expectedServiceIds) {
+ final boolean present = services.stream().anyMatch(service ->
serviceId.equals(service.getId()));
+ assertTrue(present, "Expected controller service " + serviceId + "
to be present. Services: " + describeControllerServices(services));
+ }
+ }
+
+ private void assertLocalModificationsContainComponents(final String
processGroupId, final String... componentIds)
+ throws NiFiClientException, IOException {
+ final FlowComparisonEntity localModifications =
getNifiClient().getProcessGroupClient().getLocalModifications(processGroupId);
+ assertFalse(localModifications.getComponentDifferences().isEmpty(),
+ "Expected preserved local changes to remain visible after
rebase, but none were found");
+
+ for (final String componentId : componentIds) {
+ final boolean reported =
localModifications.getComponentDifferences().stream()
+ .anyMatch(component ->
componentId.equals(component.getComponentId()));
+ assertTrue(reported, "Expected local modifications to include
component " + componentId + ". Reported differences: "
+ +
describeComponentDifferences(localModifications.getComponentDifferences()));
+ }
+ }
+
+ private String describeControllerServices(final
Collection<ControllerServiceEntity> services) {
+ final StringBuilder sb = new StringBuilder();
+ for (final ControllerServiceEntity service : services) {
+ if (!sb.isEmpty()) {
+ sb.append(", ");
+ }
+
sb.append(service.getId()).append("=").append(service.getComponent().getType());
+ }
+ return sb.length() == 0 ? "none" : sb.toString();
+ }
+
+ private String describeComponentDifferences(final
Collection<ComponentDifferenceDTO> componentDifferences) {
+ final StringBuilder sb = new StringBuilder();
+ for (final ComponentDifferenceDTO component : componentDifferences) {
+ if (!sb.isEmpty()) {
+ sb.append("; ");
+ }
+
+ sb.append(component.getComponentType()).append(" ")
+ .append(component.getComponentName()).append(" (")
+ .append(component.getComponentId()).append(")");
+
+ if (component.getDifferences() != null &&
!component.getDifferences().isEmpty()) {
+ sb.append(" -> ");
+ boolean firstDifference = true;
+ for (final DifferenceDTO difference :
component.getDifferences()) {
+ if (!firstDifference) {
+ sb.append(", ");
+ }
+ sb.append(difference.getDifference());
+ firstDifference = false;
+ }
+ }
+ }
+ return sb.length() == 0 ? "none" : sb.toString();
+ }
+
private void executeRebaseWithFingerprint(final ProcessGroupEntity group,
final String targetVersion, final String fingerprint)
throws NiFiClientException, IOException, InterruptedException {
