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 000000000000..1ba799ae052e --- /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,109 @@ +/* + * 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 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())); + } + + controllerService.setGroupIdentifier(parentGroup.getIdentifier()); + return RebaseAnalysis.ClassifiedDifference.compatible(localDifference); + } + + @Override + public void apply(final FlowDifference localDifference, final VersionedProcessGroup mergedFlow) { + final VersionedControllerService controllerService = (VersionedControllerService) localDifference.getComponentB(); + final VersionedProcessGroup parentGroup = RebaseHandlerUtils.findProcessGroupById(mergedFlow, controllerService.getGroupIdentifier()); + if (parentGroup == null) { + throw new IllegalStateException("Parent Process Group %s for Controller Service %s was verified during classification but is absent during apply" + .formatted(controllerService.getGroupIdentifier(), controllerService.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 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 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/RebaseConflictCode.java b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseConflictCode.java index 311ced78b336..ad651b6d017f 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/RebaseHandlerUtils.java b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/RebaseHandlerUtils.java index f870ea4b8a86..c1d2702d5c6a 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 @@ static VersionedComponent findComponentById(final VersionedProcessGroup group, f 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 b942d2bfccf5..c2f5fdabe48b 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 StandardRebaseEngine() { registerHandler(new PositionChangedRebaseHandler()); registerHandler(new SizeChangedRebaseHandler()); registerHandler(new BendpointsChangedRebaseHandler()); + registerHandler(new ComponentAddedRebaseHandler()); registerHandler(new PropertyChangedRebaseHandler()); registerHandler(new PropertyAddedRebaseHandler()); registerHandler(new CommentsChangedRebaseHandler()); 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 000000000000..d8424c955cc7 --- /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,292 @@ +/* + * 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 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); + + final RebaseAnalysis.ClassifiedDifference result = handler.classify(localDifference, Collections.emptySet(), targetSnapshot); + + assertEquals(RebaseClassification.COMPATIBLE, result.getClassification()); + } + + @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 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(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(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 properties = new HashMap<>(); + properties.put(SERVICE_ENABLED_PROPERTY, SERVICE_ENABLED_VALUE); + service.setProperties(properties); + + final Map 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 74e929932726..21cbac0e9a34 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 @@ 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 @@ void testConflictingPropertyChangeOnSamePropertyAndComponent() { } @Test - void testUnsupportedDifferenceTypeNoHandler() { + void testUnsupportedLocalProcessorAdditionUsesRegisteredComponentAddedHandler() { final VersionedProcessor processor = createProcessor("proc-a", "ProcessorA"); final Set localDifferences = new HashSet<>(); @@ -159,8 +183,165 @@ void testUnsupportedDifferenceTypeNoHandler() { 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 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 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.setInstanceIdentifier(VERSION_N_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 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 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 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 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 +833,28 @@ private VersionedConnection createConnection(final String identifier) { 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 +869,64 @@ private VersionedProcessor findProcessorById(final VersionedProcessGroup group, } 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 000000000000..000675731c52 --- /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 99b5498bfe44..21d068f18196 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 c698dae31bb6..e1ad570c161c 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 @@ 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.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,18 @@ 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"; @Test public void testCleanRebaseWithPositionAndPropertyChanges() throws NiFiClientException, IOException, InterruptedException { @@ -396,6 +413,98 @@ public void testRebasePreservesLocalModificationsAgainstTargetVersion() throws N "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(), "2"); + 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(), "2"); + + final VersionControlInformationDTO updatedVci = getVersionControlInfo(originalGroup.getId()); + assertEquals("2", updatedVci.getVersion()); + + final Set 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(), "2"); + 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(), "2"); + + final VersionControlInformationDTO updatedVci = getVersionControlInfo(originalGroup.getId()); + assertEquals("2", updatedVci.getVersion()); + + final Set 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 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 +543,74 @@ private ProcessorEntity findProcessorByType(final String processGroupId, final S .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 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 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 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 {