From 1440275e54669e096e438f900717653cfed11630 Mon Sep 17 00:00:00 2001 From: Pierre Villard Date: Thu, 20 Aug 2026 17:41:12 +0200 Subject: [PATCH 1/2] NIFI-16235 - Support rebasing locally added Controller Services in versioned Process Groups --- .../diff/ComponentAddedRebaseHandler.java | 107 +++++++ .../flow/diff/RebaseConflictCode.java | 10 + .../flow/diff/RebaseHandlerUtils.java | 15 + .../flow/diff/StandardRebaseEngine.java | 1 + .../diff/ComponentAddedRebaseHandlerTest.java | 284 ++++++++++++++++++ .../registry/flow/diff/RebaseEngineTest.java | 246 ++++++++++++++- ...akeDynamicPropertiesControllerService.java | 46 +++ ...g.apache.nifi.controller.ControllerService | 1 + .../system/registry/RebaseVersionIT.java | 165 ++++++++++ 9 files changed, 873 insertions(+), 2 deletions(-) create mode 100644 nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/ComponentAddedRebaseHandler.java create mode 100644 nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/ComponentAddedRebaseHandlerTest.java create mode 100644 nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions-services/src/main/java/org/apache/nifi/cs/tests/system/FakeDynamicPropertiesControllerService.java 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..5d6a132d0e8f --- /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,107 @@ +/* + * 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 { + + @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" : 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..c44c82a266e4 --- /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,284 @@ +/* + * 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 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 Controller Service", insertedService.getName()); + assertEquals("org.apache.nifi.services.LocalControllerService", insertedService.getType()); + assertEquals("group", insertedService.getBundle().getGroup()); + assertEquals("artifact", insertedService.getBundle().getArtifact()); + assertEquals("1.0.0", 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 Controller Service"); + service.setType("org.apache.nifi.services.LocalControllerService"); + service.setBundle(new Bundle("group", "artifact", "1.0.0")); + service.setScheduledState(ScheduledState.DISABLED); + service.setComments("local comments"); + + final Map properties = new HashMap<>(); + properties.put("service.enabled", "true"); + service.setProperties(properties); + + final Map descriptors = new HashMap<>(); + descriptors.put("service.enabled", createDescriptor("service.enabled", 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..59f9239bed25 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; @@ -142,7 +145,7 @@ void testConflictingPropertyChangeOnSamePropertyAndComponent() { } @Test - void testUnsupportedDifferenceTypeNoHandler() { + void testUnsupportedLocalProcessorAdditionUsesRegisteredComponentAddedHandler() { final VersionedProcessor processor = createProcessor("proc-a", "ProcessorA"); final Set localDifferences = new HashSet<>(); @@ -159,10 +162,167 @@ 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", "Service X", "root"); + final VersionedControllerService localAddedService = createControllerService("service-y", "Service Y", "root"); + + final VersionedProcessor versionNProcessor = createProcessorWithProperty("proc-a", "ProcessorA", "controller.service", "service-x"); + final VersionedProcessor localProcessor = createProcessorWithProperty("proc-a", "ProcessorA", "controller.service", "service-y"); + + 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", + "service-x", "service-y", "Processor property changed locally")); + + final VersionedProcessGroup targetSnapshot = new VersionedProcessGroup(); + targetSnapshot.setIdentifier("root"); + targetSnapshot.getControllerServices().add(versionNService); + targetSnapshot.getProcessors().add(createProcessorWithProperty("proc-a", "ProcessorA", "controller.service", "service-x")); + + 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", RebaseClassification.COMPATIBLE, null); + assertClassification(analysis, DifferenceType.PROPERTY_CHANGED, "proc-a", RebaseClassification.COMPATIBLE, null); + + final VersionedProcessGroup merged = analysis.getMergedSnapshot(); + assertNotNull(merged); + final VersionedControllerService mergedService = findControllerServiceById(merged, "service-y"); + assertSame(localAddedService, mergedService); + + final VersionedProcessor mergedProcessor = findProcessorById(merged, "proc-a"); + assertNotNull(mergedProcessor); + assertEquals("service-y", mergedProcessor.getProperties().get("controller.service")); + } + + @Test + void testAnalyzePreservesRootAdditionWhenTargetRootIdentifierChangedUpstream() { + final VersionedControllerService localAddedService = createControllerService("service-y", "Service Y", "root-n"); + 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("root-n-plus-one"); + targetRoot.setInstanceIdentifier("root-n"); + targetRoot.setName("Root"); + + final RebaseAnalysis analysis = engine.analyze(localDifferences, Collections.emptySet(), targetRoot); + + assertTrue(analysis.isRebaseAllowed()); + assertSame(localAddedService, findControllerServiceById(targetRoot, "service-y")); + assertEquals("root-n-plus-one", localAddedService.getGroupIdentifier()); + } + + @Test + void testAnalyzeRejectsRootAdditionWhenParentRemovedUpstream() { + final VersionedControllerService localAddedService = createControllerService("service-y", "Service Y", "root-n"); + 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("root-n"); + + 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", RebaseClassification.UNSUPPORTED, + RebaseConflictCode.COMPONENT_NOT_FOUND); + } + + @Test + void testAnalyzeScenario2PreservesMultipleAddedControllerServicesAndDynamicReferences() { + final VersionedControllerService versionNServiceA = createControllerService("service-a", "Service A", "root"); + versionNServiceA.setProperties(Collections.emptyMap()); + versionNServiceA.setPropertyDescriptors(Collections.emptyMap()); + + final VersionedControllerService localServiceA = createControllerService("service-a", "Service A", "root"); + localServiceA.setProperties(Map.of("dynamic.y", "service-y", "dynamic.z", "service-z")); + localServiceA.setPropertyDescriptors(Map.of( + "dynamic.y", createPropertyDescriptor("dynamic.y", true, false), + "dynamic.z", createPropertyDescriptor("dynamic.z", true, false))); + + final VersionedControllerService localServiceY = createControllerService("service-y", "Service Y", "root"); + final VersionedControllerService localServiceZ = createControllerService("service-z", "Service Z", "root"); + + 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", + null, "service-y", "Dynamic property added for service Y")); + localDifferences.add(new StandardFlowDifference(DifferenceType.PROPERTY_ADDED, versionNServiceA, localServiceA, "dynamic.z", + null, "service-z", "Dynamic property added for service Z")); + + final VersionedProcessGroup targetSnapshot = new VersionedProcessGroup(); + targetSnapshot.setIdentifier("root"); + 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", RebaseClassification.COMPATIBLE, null); + assertClassification(analysis, DifferenceType.COMPONENT_ADDED, "service-z", RebaseClassification.COMPATIBLE, null); + + final VersionedProcessGroup merged = analysis.getMergedSnapshot(); + assertNotNull(merged); + assertSame(localServiceY, findControllerServiceById(merged, "service-y")); + assertSame(localServiceZ, findControllerServiceById(merged, "service-z")); + + final VersionedControllerService mergedServiceA = findControllerServiceById(merged, "service-a"); + assertNotNull(mergedServiceA); + assertEquals("service-y", mergedServiceA.getProperties().get("dynamic.y")); + assertEquals("service-z", mergedServiceA.getProperties().get("dynamic.z")); + } + + @Test + void testAnalyzeCollisionBlocksRebaseAndDoesNotMutateTargetSnapshot() { + final VersionedControllerService collidingTargetService = createControllerService("service-y", "Target Service Y", "root"); + final VersionedControllerService localAddedService = createControllerService("service-y", "Local Service Y", "root"); + + final VersionedProcessor versionNProcessor = createProcessorWithProperty("proc-a", "ProcessorA", "controller.service", "service-x"); + final VersionedProcessor localProcessor = createProcessorWithProperty("proc-a", "ProcessorA", "controller.service", "service-y"); + + 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", + "service-x", "service-y", "Processor property changed locally")); + + final VersionedProcessGroup targetSnapshot = new VersionedProcessGroup(); + targetSnapshot.setIdentifier("root"); + targetSnapshot.getControllerServices().add(collidingTargetService); + targetSnapshot.getProcessors().add(createProcessorWithProperty("proc-a", "ProcessorA", "controller.service", "service-x")); + + final RebaseAnalysis analysis = engine.analyze(localDifferences, Collections.emptySet(), targetSnapshot); + + assertFalse(analysis.isRebaseAllowed()); + assertNull(analysis.getMergedSnapshot()); + assertClassification(analysis, DifferenceType.COMPONENT_ADDED, "service-y", RebaseClassification.CONFLICTING, + RebaseConflictCode.COMPONENT_IDENTIFIER_COLLISION); + assertEquals(1, countComponentsById(targetSnapshot, "service-y")); + + final VersionedProcessor unchangedProcessor = findProcessorById(targetSnapshot, "proc-a"); + assertNotNull(unchangedProcessor); + assertEquals("service-x", unchangedProcessor.getProperties().get("controller.service")); + } + @Test void testMixedCompatibleAndUnsupported() { final VersionedProcessor processorA = createProcessor("proc-a", "ProcessorA"); @@ -652,6 +812,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("org.apache.nifi.services.LocalControllerService"); + service.setBundle(new Bundle("group", "artifact", "1.0.0")); + 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 +848,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..a715059f5fe1 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; @@ -396,6 +401,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", "root"); + final ControllerServiceEntity serviceX = util.createControllerService("FakeControllerService1", originalGroup.getId()); + final ProcessorEntity fakeProcessor = util.createProcessor("FakeProcessor", originalGroup.getId()); + util.updateProcessorProperties(fakeProcessor, Map.of("Fake Service", serviceX.getId())); + util.createProcessor("GenerateFlowFile", 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", clientEntity.getId(), TEST_FLOWS_BUCKET, flowId, "1"); + final ProcessorEntity upstreamGenerate = findProcessorByType(secondGroup.getId(), "GenerateFlowFile"); + util.updateProcessorProperties(upstreamGenerate, Map.of("Text", "upstream-change")); + util.saveFlowVersion(secondGroup, clientEntity, getVersionControlInformation(secondGroup.getId())); + + final ControllerServiceEntity serviceY = util.createControllerService("FakeControllerService1", originalGroup.getId()); + util.updateProcessorProperties(fakeProcessor, Map.of("Fake Service", 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(), "FakeProcessor"); + assertEquals(serviceY.getId(), rebasedProcessor.getComponent().getConfig().getProperties().get("Fake Service")); + + 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", "root"); + final ControllerServiceEntity serviceX = util.createControllerService("FakeControllerService1", originalGroup.getId()); + final ControllerServiceEntity dynamicService = util.createControllerService("FakeDynamicPropertiesControllerService", originalGroup.getId()); + util.updateControllerServiceProperties(dynamicService, Collections.singletonMap("FCS.X", serviceX.getId())); + util.createProcessor("GenerateFlowFile", 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", clientEntity.getId(), TEST_FLOWS_BUCKET, flowId, "1"); + final ProcessorEntity upstreamGenerate = findProcessorByType(secondGroup.getId(), "GenerateFlowFile"); + util.updateProcessorProperties(upstreamGenerate, Map.of("Text", "upstream-change")); + util.saveFlowVersion(secondGroup, clientEntity, getVersionControlInformation(secondGroup.getId())); + + final ControllerServiceEntity serviceY = util.createControllerService("FakeControllerService1", originalGroup.getId()); + final ControllerServiceEntity serviceZ = util.createControllerService("FakeControllerService1", originalGroup.getId()); + util.updateControllerServiceProperties(dynamicService, Map.of( + "FCS.X", serviceX.getId(), + "FCS.Y", serviceY.getId(), + "FCS.Z", 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("FCS.X")); + assertEquals(serviceY.getId(), dynamicProperties.get("FCS.Y")); + assertEquals(serviceZ.getId(), dynamicProperties.get("FCS.Z")); + + 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 +531,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 { From dcebaf3cae15cab7418b627b9da0a7bbdf2339c1 Mon Sep 17 00:00:00 2001 From: Pierre Villard Date: Wed, 26 Aug 2026 20:52:52 +0200 Subject: [PATCH 2/2] review --- .../diff/ComponentAddedRebaseHandler.java | 10 +- .../diff/ComponentAddedRebaseHandlerTest.java | 34 +++-- .../registry/flow/diff/RebaseEngineTest.java | 129 ++++++++++-------- .../system/registry/RebaseVersionIT.java | 68 +++++---- 4 files changed, 142 insertions(+), 99 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 index 5d6a132d0e8f..1ba799ae052e 100644 --- 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 @@ -26,6 +26,8 @@ public class ComponentAddedRebaseHandler implements RebaseHandler { + private static final String NULL_COMPONENT_TYPE = "null"; + @Override public DifferenceType getSupportedType() { return DifferenceType.COMPONENT_ADDED; @@ -36,7 +38,7 @@ public RebaseAnalysis.ClassifiedDifference classify(final FlowDifference localDi final VersionedProcessGroup targetSnapshot) { final VersionedComponent addedComponent = localDifference.getComponentB(); if (!(addedComponent instanceof VersionedControllerService controllerService)) { - final String componentType = addedComponent == null ? "null" : addedComponent.getClass().getSimpleName(); + 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)); } @@ -44,13 +46,13 @@ public RebaseAnalysis.ClassifiedDifference classify(final FlowDifference localDi 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())); + "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" + "Parent Process Group %s for Controller Service %s not found in target snapshot" .formatted(parentGroupIdentifier, controllerService.getIdentifier())); } @@ -70,7 +72,7 @@ public void apply(final FlowDifference localDifference, final VersionedProcessGr 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" + 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())); } 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 index c44c82a266e4..d8424c955cc7 100644 --- 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 @@ -45,6 +45,14 @@ class ComponentAddedRebaseHandlerTest { 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; @@ -165,14 +173,14 @@ void testApplyAddsControllerServiceToRootPreservingIdentityAndConfiguration() { assertSame(addedService, insertedService); assertEquals(ADDED_SERVICE_ID, insertedService.getIdentifier()); assertEquals(ROOT, insertedService.getGroupIdentifier()); - assertEquals("Local Controller Service", insertedService.getName()); - assertEquals("org.apache.nifi.services.LocalControllerService", insertedService.getType()); - assertEquals("group", insertedService.getBundle().getGroup()); - assertEquals("artifact", insertedService.getBundle().getArtifact()); - assertEquals("1.0.0", insertedService.getBundle().getVersion()); + 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(LOCAL_COMMENTS, insertedService.getComments()); assertEquals(ScheduledState.DISABLED, insertedService.getScheduledState()); } @@ -204,7 +212,7 @@ void testApplyThrowsWhenVerifiedParentIsMissing() { 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", + assertEquals("Parent Process Group child for Controller Service service-y was verified during classification but is absent during apply", exception.getMessage()); } @@ -258,18 +266,18 @@ private VersionedControllerService createControllerService(final String identifi final VersionedControllerService service = new VersionedControllerService(); service.setIdentifier(identifier); service.setGroupIdentifier(groupIdentifier); - service.setName("Local Controller Service"); - service.setType("org.apache.nifi.services.LocalControllerService"); - service.setBundle(new Bundle("group", "artifact", "1.0.0")); + 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"); + service.setComments(LOCAL_COMMENTS); final Map properties = new HashMap<>(); - properties.put("service.enabled", "true"); + properties.put(SERVICE_ENABLED_PROPERTY, SERVICE_ENABLED_VALUE); service.setProperties(properties); final Map descriptors = new HashMap<>(); - descriptors.put("service.enabled", createDescriptor("service.enabled", false, false)); + descriptors.put(SERVICE_ENABLED_PROPERTY, createDescriptor(SERVICE_ENABLED_PROPERTY, false, false)); service.setPropertyDescriptors(descriptors); return service; } 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 59f9239bed25..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 @@ -46,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 @@ -168,67 +189,67 @@ void testUnsupportedLocalProcessorAdditionUsesRegisteredComponentAddedHandler() @Test void testAnalyzeScenario1PreservesAddedControllerServiceAndProcessorReference() { - final VersionedControllerService versionNService = createControllerService("service-x", "Service X", "root"); - final VersionedControllerService localAddedService = createControllerService("service-y", "Service Y", "root"); + 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("proc-a", "ProcessorA", "controller.service", "service-x"); - final VersionedProcessor localProcessor = createProcessorWithProperty("proc-a", "ProcessorA", "controller.service", "service-y"); + 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", - "service-x", "service-y", "Processor property changed 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"); + targetSnapshot.setIdentifier(ROOT_ID); targetSnapshot.getControllerServices().add(versionNService); - targetSnapshot.getProcessors().add(createProcessorWithProperty("proc-a", "ProcessorA", "controller.service", "service-x")); + 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", RebaseClassification.COMPATIBLE, null); - assertClassification(analysis, DifferenceType.PROPERTY_CHANGED, "proc-a", RebaseClassification.COMPATIBLE, null); + 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"); + final VersionedControllerService mergedService = findControllerServiceById(merged, SERVICE_Y_ID); assertSame(localAddedService, mergedService); - final VersionedProcessor mergedProcessor = findProcessorById(merged, "proc-a"); + final VersionedProcessor mergedProcessor = findProcessorById(merged, PROCESSOR_ID); assertNotNull(mergedProcessor); - assertEquals("service-y", mergedProcessor.getProperties().get("controller.service")); + assertEquals(SERVICE_Y_ID, mergedProcessor.getProperties().get(CONTROLLER_SERVICE_PROPERTY)); } @Test void testAnalyzePreservesRootAdditionWhenTargetRootIdentifierChangedUpstream() { - final VersionedControllerService localAddedService = createControllerService("service-y", "Service Y", "root-n"); + 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("root-n-plus-one"); - targetRoot.setInstanceIdentifier("root-n"); + 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")); - assertEquals("root-n-plus-one", localAddedService.getGroupIdentifier()); + assertSame(localAddedService, findControllerServiceById(targetRoot, SERVICE_Y_ID)); + assertEquals(TARGET_ROOT_ID, localAddedService.getGroupIdentifier()); } @Test void testAnalyzeRejectsRootAdditionWhenParentRemovedUpstream() { - final VersionedControllerService localAddedService = createControllerService("service-y", "Service Y", "root-n"); + 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("root-n"); + removedParent.setIdentifier(VERSION_N_ROOT_ID); final VersionedProcessGroup targetRoot = new VersionedProcessGroup(); targetRoot.setIdentifier("replacement-root"); @@ -239,37 +260,37 @@ void testAnalyzeRejectsRootAdditionWhenParentRemovedUpstream() { final RebaseAnalysis analysis = engine.analyze(localDifferences, upstreamDifferences, targetRoot); assertFalse(analysis.isRebaseAllowed()); - assertClassification(analysis, DifferenceType.COMPONENT_ADDED, "service-y", RebaseClassification.UNSUPPORTED, + assertClassification(analysis, DifferenceType.COMPONENT_ADDED, SERVICE_Y_ID, RebaseClassification.UNSUPPORTED, RebaseConflictCode.COMPONENT_NOT_FOUND); } @Test void testAnalyzeScenario2PreservesMultipleAddedControllerServicesAndDynamicReferences() { - final VersionedControllerService versionNServiceA = createControllerService("service-a", "Service A", "root"); + 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", "Service A", "root"); - localServiceA.setProperties(Map.of("dynamic.y", "service-y", "dynamic.z", "service-z")); + 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", createPropertyDescriptor("dynamic.y", true, false), - "dynamic.z", createPropertyDescriptor("dynamic.z", true, false))); + DYNAMIC_Y_PROPERTY, createPropertyDescriptor(DYNAMIC_Y_PROPERTY, true, false), + DYNAMIC_Z_PROPERTY, createPropertyDescriptor(DYNAMIC_Z_PROPERTY, true, false))); - final VersionedControllerService localServiceY = createControllerService("service-y", "Service Y", "root"); - final VersionedControllerService localServiceZ = createControllerService("service-z", "Service Z", "root"); + 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", - null, "service-y", "Dynamic property added for service Y")); - localDifferences.add(new StandardFlowDifference(DifferenceType.PROPERTY_ADDED, versionNServiceA, localServiceA, "dynamic.z", - null, "service-z", "Dynamic property added for service Z")); + 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"); + targetSnapshot.setIdentifier(ROOT_ID); targetSnapshot.getControllerServices().add(versionNServiceA); final RebaseAnalysis analysis = engine.analyze(localDifferences, Collections.emptySet(), targetSnapshot); @@ -277,50 +298,50 @@ void testAnalyzeScenario2PreservesMultipleAddedControllerServicesAndDynamicRefer assertTrue(analysis.isRebaseAllowed()); assertEquals(4, analysis.getClassifiedLocalChanges().size()); assertAllCompatible(analysis); - assertClassification(analysis, DifferenceType.COMPONENT_ADDED, "service-y", RebaseClassification.COMPATIBLE, null); - assertClassification(analysis, DifferenceType.COMPONENT_ADDED, "service-z", RebaseClassification.COMPATIBLE, null); + 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")); - assertSame(localServiceZ, findControllerServiceById(merged, "service-z")); + assertSame(localServiceY, findControllerServiceById(merged, SERVICE_Y_ID)); + assertSame(localServiceZ, findControllerServiceById(merged, SERVICE_Z_ID)); - final VersionedControllerService mergedServiceA = findControllerServiceById(merged, "service-a"); + final VersionedControllerService mergedServiceA = findControllerServiceById(merged, SERVICE_A_ID); assertNotNull(mergedServiceA); - assertEquals("service-y", mergedServiceA.getProperties().get("dynamic.y")); - assertEquals("service-z", mergedServiceA.getProperties().get("dynamic.z")); + 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", "Target Service Y", "root"); - final VersionedControllerService localAddedService = createControllerService("service-y", "Local Service Y", "root"); + 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("proc-a", "ProcessorA", "controller.service", "service-x"); - final VersionedProcessor localProcessor = createProcessorWithProperty("proc-a", "ProcessorA", "controller.service", "service-y"); + 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", - "service-x", "service-y", "Processor property changed 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"); + targetSnapshot.setIdentifier(ROOT_ID); targetSnapshot.getControllerServices().add(collidingTargetService); - targetSnapshot.getProcessors().add(createProcessorWithProperty("proc-a", "ProcessorA", "controller.service", "service-x")); + 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", RebaseClassification.CONFLICTING, + assertClassification(analysis, DifferenceType.COMPONENT_ADDED, SERVICE_Y_ID, RebaseClassification.CONFLICTING, RebaseConflictCode.COMPONENT_IDENTIFIER_COLLISION); - assertEquals(1, countComponentsById(targetSnapshot, "service-y")); + assertEquals(1, countComponentsById(targetSnapshot, SERVICE_Y_ID)); - final VersionedProcessor unchangedProcessor = findProcessorById(targetSnapshot, "proc-a"); + final VersionedProcessor unchangedProcessor = findProcessorById(targetSnapshot, PROCESSOR_ID); assertNotNull(unchangedProcessor); - assertEquals("service-x", unchangedProcessor.getProperties().get("controller.service")); + assertEquals(SERVICE_X_ID, unchangedProcessor.getProperties().get(CONTROLLER_SERVICE_PROPERTY)); } @Test @@ -817,8 +838,8 @@ private VersionedControllerService createControllerService(final String identifi service.setIdentifier(identifier); service.setName(name); service.setGroupIdentifier(groupIdentifier); - service.setType("org.apache.nifi.services.LocalControllerService"); - service.setBundle(new Bundle("group", "artifact", "1.0.0")); + 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()); 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 a715059f5fe1..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 @@ -49,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 { @@ -406,23 +418,23 @@ public void testRebasePreservesLocallyAddedControllerServiceReferencedByProcesso final FlowRegistryClientEntity clientEntity = registerClient(); final NiFiClientUtil util = getClientUtil(); - final ProcessGroupEntity originalGroup = util.createProcessGroup("Original", "root"); - final ControllerServiceEntity serviceX = util.createControllerService("FakeControllerService1", originalGroup.getId()); - final ProcessorEntity fakeProcessor = util.createProcessor("FakeProcessor", originalGroup.getId()); - util.updateProcessorProperties(fakeProcessor, Map.of("Fake Service", serviceX.getId())); - util.createProcessor("GenerateFlowFile", originalGroup.getId()); + 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", clientEntity.getId(), TEST_FLOWS_BUCKET, flowId, "1"); - final ProcessorEntity upstreamGenerate = findProcessorByType(secondGroup.getId(), "GenerateFlowFile"); - util.updateProcessorProperties(upstreamGenerate, Map.of("Text", "upstream-change")); + 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("FakeControllerService1", originalGroup.getId()); - util.updateProcessorProperties(fakeProcessor, Map.of("Fake Service", serviceY.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() @@ -437,8 +449,8 @@ public void testRebasePreservesLocallyAddedControllerServiceReferencedByProcesso final Set rebasedServices = getNifiClient().getFlowClient().getControllerServices(originalGroup.getId()).getControllerServices(); assertControllerServicesPresent(rebasedServices, serviceX.getId(), serviceY.getId()); - final ProcessorEntity rebasedProcessor = findProcessorByType(originalGroup.getId(), "FakeProcessor"); - assertEquals(serviceY.getId(), rebasedProcessor.getComponent().getConfig().getProperties().get("Fake Service")); + 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()); } @@ -449,27 +461,27 @@ public void testRebasePreservesLocallyAddedControllerServicesReferencedByDynamic final FlowRegistryClientEntity clientEntity = registerClient(); final NiFiClientUtil util = getClientUtil(); - final ProcessGroupEntity originalGroup = util.createProcessGroup("Original", "root"); - final ControllerServiceEntity serviceX = util.createControllerService("FakeControllerService1", originalGroup.getId()); - final ControllerServiceEntity dynamicService = util.createControllerService("FakeDynamicPropertiesControllerService", originalGroup.getId()); - util.updateControllerServiceProperties(dynamicService, Collections.singletonMap("FCS.X", serviceX.getId())); - util.createProcessor("GenerateFlowFile", originalGroup.getId()); + 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", clientEntity.getId(), TEST_FLOWS_BUCKET, flowId, "1"); - final ProcessorEntity upstreamGenerate = findProcessorByType(secondGroup.getId(), "GenerateFlowFile"); - util.updateProcessorProperties(upstreamGenerate, Map.of("Text", "upstream-change")); + 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("FakeControllerService1", originalGroup.getId()); - final ControllerServiceEntity serviceZ = util.createControllerService("FakeControllerService1", originalGroup.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( - "FCS.X", serviceX.getId(), - "FCS.Y", serviceY.getId(), - "FCS.Z", serviceZ.getId())); + 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() @@ -486,9 +498,9 @@ public void testRebasePreservesLocallyAddedControllerServicesReferencedByDynamic final ControllerServiceEntity rebasedDynamicService = getNifiClient().getControllerServicesClient().getControllerService(dynamicService.getId()); final Map dynamicProperties = rebasedDynamicService.getComponent().getProperties(); - assertEquals(serviceX.getId(), dynamicProperties.get("FCS.X")); - assertEquals(serviceY.getId(), dynamicProperties.get("FCS.Y")); - assertEquals(serviceZ.getId(), dynamicProperties.get("FCS.Z")); + 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()); }