NIFI-16058 - Treat external controller service reference changes in versioned flows as environment-specific

nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java

@@ -68,13 +68,9 @@
 import org.apache.nifi.controller.service.StandardConfigurationContext;
 import org.apache.nifi.encrypt.PropertyEncryptor;
 import org.apache.nifi.flow.ExecutionEngine;
-import org.apache.nifi.flow.ExternalControllerServiceReference;
 import org.apache.nifi.flow.VersionedComponent;
-import org.apache.nifi.flow.VersionedControllerService;
 import org.apache.nifi.flow.VersionedExternalFlow;
 import org.apache.nifi.flow.VersionedProcessGroup;
-import org.apache.nifi.flow.VersionedProcessor;
-import org.apache.nifi.flow.VersionedPropertyDescriptor;
 import org.apache.nifi.flow.synchronization.StandardVersionedComponentSynchronizer;
 import org.apache.nifi.flow.synchronization.VersionedFlowSynchronizationContext;
 import org.apache.nifi.lifecycle.ProcessorStopLifecycleMethods;
@@ -3867,7 +3863,6 @@ public void synchronizeWithFlowRegistry(final FlowManager flowManager) {
                 final FlowSnapshotContainer registrySnapshotContainer = flowRegistry.getFlowContents(
                         FlowRegistryClientContextFactory.getAnonymousContext(), flowVersionLocation, false);
                 final RegisteredFlowSnapshot registrySnapshot = registrySnapshotContainer.getFlowSnapshot();
-                resolveExternalServiceReferences(registrySnapshot);
                 final VersionedProcessGroup registryFlow = registrySnapshot.getFlowContents();
                 vci.setFlowSnapshot(registryFlow);
             } catch (final IOException | FlowRegistryException e) {
@@ -4091,81 +4086,6 @@ private String generateUuid(final String proposedId, final String destinationGro
         return uuid.toString();
     }
 
-    private void resolveExternalServiceReferences(final RegisteredFlowSnapshot snapshot) {
-        final Map<String, ExternalControllerServiceReference> externalRefs = snapshot.getExternalControllerServices();
-        if (externalRefs == null || externalRefs.isEmpty()) {
-            return;
-        }
-
-        final ProcessGroup parentGroup = getParent();
-        if (parentGroup == null) {
-            return;
-        }
-
-        final Map<String, String> serviceNameToVersionedId = new HashMap<>();
-        for (final ControllerServiceNode serviceNode : parentGroup.getControllerServices(true)) {
-            final String versionedId = serviceNode.getVersionedComponentId().orElse(
-                    VersionedComponentFlowMapper.generateVersionedComponentId(serviceNode.getIdentifier()));
-            serviceNameToVersionedId.put(serviceNode.getName(), versionedId);
-        }
-
-        final Map<String, String> foreignToLocalId = new HashMap<>();
-        for (final Map.Entry<String, ExternalControllerServiceReference> entry : externalRefs.entrySet()) {
-            final String foreignId = entry.getKey();
-            final String serviceName = entry.getValue().getName();
-            final String localId = serviceNameToVersionedId.get(serviceName);
-            if (localId != null && !localId.equals(foreignId)) {
-                foreignToLocalId.put(foreignId, localId);
-            }
-        }
-
-        if (!foreignToLocalId.isEmpty()) {
-            replaceExternalServiceIds(snapshot.getFlowContents(), foreignToLocalId);
-        }
-    }
-
-    private void replaceExternalServiceIds(final VersionedProcessGroup group, final Map<String, String> foreignToLocalId) {
-        if (group.getProcessors() != null) {
-            for (final VersionedProcessor processor : group.getProcessors()) {
-                replaceServicePropertyIds(processor.getProperties(), processor.getPropertyDescriptors(), foreignToLocalId);
-            }
-        }
-
-        if (group.getControllerServices() != null) {
-            for (final VersionedControllerService service : group.getControllerServices()) {
-                replaceServicePropertyIds(service.getProperties(), service.getPropertyDescriptors(), foreignToLocalId);
-            }
-        }
-
-        if (group.getProcessGroups() != null) {
-            for (final VersionedProcessGroup child : group.getProcessGroups()) {
-                replaceExternalServiceIds(child, foreignToLocalId);
-            }
-        }
-    }
-
-    private void replaceServicePropertyIds(final Map<String, String> properties, final Map<String, VersionedPropertyDescriptor> descriptors,
-                                           final Map<String, String> foreignToLocalId) {
-        if (properties == null || descriptors == null) {
-            return;
-        }
-
-        for (final Map.Entry<String, String> entry : properties.entrySet()) {
-            final String propertyValue = entry.getValue();
-            if (propertyValue == null) {
-                continue;
-            }
-
-            final VersionedPropertyDescriptor descriptor = descriptors.get(entry.getKey());
-            if (descriptor != null && descriptor.getIdentifiesControllerService()) {
-                final String localId = foreignToLocalId.get(propertyValue);
-                if (localId != null) {
-                    entry.setValue(localId);
-                }
-            }
-        }
-    }
-
     private Set<FlowDifference> getModifications() {
         final StandardVersionControlInformation vci = versionControlInfo.get();
 
@@ -4199,7 +4119,7 @@ private Set<FlowDifference> getModifications() {
             final FlowComparison comparison = flowComparator.compare();
             final Collection<FlowDifference> comparisonDifferences = comparison.getDifferences();
             final FlowDifferenceFilters.EnvironmentalChangeContext environmentalContext =
-                FlowDifferenceFilters.buildEnvironmentalChangeContext(comparisonDifferences, flowManager);
+                FlowDifferenceFilters.buildEnvironmentalChangeContext(comparisonDifferences, versionedGroup, flowManager);
 
             final Set<FlowDifference> differences = comparisonDifferences.stream()
                 .filter(difference -> !FlowDifferenceFilters.isEnvironmentalChange(difference, versionedGroup, flowManager, environmentalContext))

nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/FlowDifferenceFilters.java

@@ -48,7 +48,9 @@
 import org.apache.nifi.registry.flow.mapping.InstantiatedVersionedComponent;
 import org.apache.nifi.registry.flow.mapping.InstantiatedVersionedConnection;
 import org.apache.nifi.registry.flow.mapping.InstantiatedVersionedControllerService;
+import org.apache.nifi.registry.flow.mapping.InstantiatedVersionedProcessGroup;
 import org.apache.nifi.registry.flow.mapping.InstantiatedVersionedProcessor;
+import org.apache.nifi.registry.flow.mapping.VersionedComponentFlowMapper;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -104,7 +106,8 @@ public static boolean isEnvironmentalChange(final FlowDifference difference, fin
             || isPropertyParameterizationRename(difference, evaluatedContext)
             || isPropertyRenameWithMatchingValue(difference, evaluatedContext)
             || isSelectedRelationshipChangeForNewRelationship(difference, flowManager)
-            || isPropertyAddedFromMigration(difference, flowManager);
+            || isPropertyAddedFromMigration(difference, flowManager)
+            || isExternalControllerServiceReferenceChange(difference, flowManager, evaluatedContext);
     }
 
     /**
@@ -706,10 +709,17 @@ private static boolean isLogFileSuffixChange(final FlowDifference flowDifference
     }
 
     public static EnvironmentalChangeContext buildEnvironmentalChangeContext(final Collection<FlowDifference> differences, final FlowManager flowManager) {
+        return buildEnvironmentalChangeContext(differences, null, flowManager);
+    }
+
+    public static EnvironmentalChangeContext buildEnvironmentalChangeContext(final Collection<FlowDifference> differences, final VersionedProcessGroup localGroup,
+                                                                             final FlowManager flowManager) {
         if (differences == null || differences.isEmpty() || flowManager == null) {
             return EnvironmentalChangeContext.empty();
         }
 
+        final Set<String> ancestorControllerServiceIds = computeAncestorControllerServiceIds(localGroup, flowManager);
+
         final Map<String, List<PropertyDiffInfo>> parameterizedAddsByComponent = new HashMap<>();
         final Map<String, List<PropertyDiffInfo>> parameterizationRemovalsByComponent = new HashMap<>();
         final Map<String, List<PropertyDiffInfo>> controllerServicePropertyAddsByValue = new HashMap<>();
@@ -855,11 +865,83 @@ && isControllerServiceProperty(difference, flowManager)) {
             }
         }
 
-        if (serviceIdsWithMatchingAdditions.isEmpty() && parameterizedPropertyRenameDifferences.isEmpty() && propertyRenamesWithMatchingValues.isEmpty()) {
+        if (serviceIdsWithMatchingAdditions.isEmpty() && parameterizedPropertyRenameDifferences.isEmpty() && propertyRenamesWithMatchingValues.isEmpty()
+                && ancestorControllerServiceIds.isEmpty()) {
             return EnvironmentalChangeContext.empty();
         }
 
-        return new EnvironmentalChangeContext(serviceIdsWithMatchingAdditions, parameterizedPropertyRenameDifferences, propertyRenamesWithMatchingValues);
+        return new EnvironmentalChangeContext(serviceIdsWithMatchingAdditions, parameterizedPropertyRenameDifferences, propertyRenamesWithMatchingValues,
+                ancestorControllerServiceIds);
+    }
+
+    /**
+     * Determines whether a Flow Difference represents a change to a property that references an <em>external</em> controller service
+     * (one defined outside the versioned Process Group, e.g. in an ancestor group). Such a reference is environment-specific: the same
+     * logical service typically has a different identifier in each environment, so pointing the property at a different external service
+     * is not treated as a local modification. The change is environmental only when the versioned-snapshot value does not resolve to a
+     * locally-accessible ancestor service while the local value does.
+     */
+    private static boolean isExternalControllerServiceReferenceChange(final FlowDifference difference, final FlowManager flowManager,
+                                                                      final EnvironmentalChangeContext context) {
+        if (difference.getDifferenceType() != DifferenceType.PROPERTY_CHANGED) {
+            return false;
+        }
+
+        final Set<String> ancestorServiceIds = context.ancestorControllerServiceIds();
+        if (ancestorServiceIds.isEmpty()) {
+            return false;
+        }
+
+        if (!isControllerServiceProperty(difference, flowManager)) {
+            return false;
+        }
+
+        final Optional<String> snapshotValue = getPropertyValue(difference, true);
+        final Optional<String> localValue = getPropertyValue(difference, false);
+        if (snapshotValue.isEmpty() || localValue.isEmpty()) {
+            return false;
+        }
+
+        // Parameter-referenced controller services are environment-portable through Parameter Contexts and are not evaluated here.
+        if (isParameterReference(snapshotValue.get()) || isParameterReference(localValue.get())) {
+            return false;
+        }
+
+        // Environmental only when the snapshot reference is not a locally-accessible ancestor controller service but the local reference
+        // is. A switch between two locally-accessible ancestor services remains a reported change, and a reference to a service defined
+        // inside the versioned Process Group is never in the ancestor set, so it always remains a reported change.
+        final boolean snapshotAccessible = ancestorServiceIds.contains(snapshotValue.get());
+        final boolean localAccessible = ancestorServiceIds.contains(localValue.get());
+        return !snapshotAccessible && localAccessible;
+    }
+
+    /**
+     * Computes the set of versioned component identifiers for all controller services accessible from the ancestors of the given (local)
+     * versioned Process Group -- i.e. services defined in the parent group and above. These identify references to external controller
+     * services. Returns an empty set when the group is not an instantiated (live-backed) group, so callers that pass a plain snapshot
+     * group get no suppression.
+     */
+    private static Set<String> computeAncestorControllerServiceIds(final VersionedProcessGroup localGroup, final FlowManager flowManager) {
+        if (flowManager == null || !(localGroup instanceof InstantiatedVersionedProcessGroup instantiatedGroup)) {
+            return Collections.emptySet();
+        }
+
+        final ProcessGroup liveGroup = flowManager.getGroup(instantiatedGroup.getInstanceIdentifier());
+        if (liveGroup == null) {
+            return Collections.emptySet();
+        }
+
+        final ProcessGroup parentGroup = liveGroup.getParent();
+        if (parentGroup == null) {
+            return Collections.emptySet();
+        }
+
+        final Set<String> ancestorServiceIds = new HashSet<>();
+        for (final ControllerServiceNode serviceNode : parentGroup.getControllerServices(true)) {
+            ancestorServiceIds.add(serviceNode.getVersionedComponentId().orElseGet(
+                    () -> VersionedComponentFlowMapper.generateVersionedComponentId(serviceNode.getIdentifier())));
+        }
+        return ancestorServiceIds;
     }
 
     public static boolean isControllerServiceCreatedForNewProperty(final FlowDifference difference, final EnvironmentalChangeContext context) {
@@ -1212,18 +1294,22 @@ FlowDifference difference() {
     }
 
     public static final class EnvironmentalChangeContext {
-        private static final EnvironmentalChangeContext EMPTY = new EnvironmentalChangeContext(Collections.emptySet(), Collections.emptySet(), Collections.emptySet());
+        private static final EnvironmentalChangeContext EMPTY =
+                new EnvironmentalChangeContext(Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), Collections.emptySet());
 
         private final Set<String> serviceIdsCreatedForNewProperties;
         private final Set<FlowDifference> parameterizedPropertyRenames;
         private final Set<FlowDifference> propertyRenamesWithMatchingValues;
+        private final Set<String> ancestorControllerServiceIds;
 
         private EnvironmentalChangeContext(final Set<String> serviceIdsCreatedForNewProperties,
                                            final Set<FlowDifference> parameterizedPropertyRenames,
-                                           final Set<FlowDifference> propertyRenamesWithMatchingValues) {
+                                           final Set<FlowDifference> propertyRenamesWithMatchingValues,
+                                           final Set<String> ancestorControllerServiceIds) {
             this.serviceIdsCreatedForNewProperties = Collections.unmodifiableSet(new HashSet<>(serviceIdsCreatedForNewProperties));
             this.parameterizedPropertyRenames = Collections.unmodifiableSet(new HashSet<>(parameterizedPropertyRenames));
             this.propertyRenamesWithMatchingValues = Collections.unmodifiableSet(new HashSet<>(propertyRenamesWithMatchingValues));
+            this.ancestorControllerServiceIds = Collections.unmodifiableSet(new HashSet<>(ancestorControllerServiceIds));
         }
 
         static EnvironmentalChangeContext empty() {
@@ -1241,5 +1327,9 @@ Set<FlowDifference> parameterizedPropertyRenames() {
         Set<FlowDifference> propertyRenamesWithMatchingValues() {
             return propertyRenamesWithMatchingValues;
         }
+
+        Set<String> ancestorControllerServiceIds() {
+            return ancestorControllerServiceIds;
+        }
     }
 }