NIFI-15016 Ensure nested versioned groups detect scheduled-state changes (#10356)

Signed-off-by: David Handermann <[email protected]>

Author: pvillard31

nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardFlowComparator.java

@@ -564,8 +564,8 @@ private void compare(final VersionedProcessGroup groupA, final VersionedProcessG
         // - explicitly requested comparison for embedded versioned groups
         final boolean shouldCompareVersioned = flowCoordinateDifferences.stream()
             .anyMatch(diff -> !diff.getFieldName().isPresent() || !diff.getFieldName().get().equals(FLOW_VERSION)) || flowComparatorVersionedStrategy == FlowComparatorVersionedStrategy.DEEP;
-        final boolean compareGroupContents = (groupACoordinates == null && groupBCoordinates == null)
-            || (groupACoordinates != null && groupBCoordinates != null && shouldCompareVersioned);
+        final boolean bothGroupsVersioned = groupACoordinates != null && groupBCoordinates != null;
+        final boolean compareGroupContents = !bothGroupsVersioned || shouldCompareVersioned || hasProcessGroupContents(groupA) || hasProcessGroupContents(groupB);
 
 
         if (compareGroupContents) {
@@ -620,6 +620,22 @@ private void extractPGComponentsDifferences(final VersionedProcessGroup groupA,
                 this::compare));
     }
 
+    private boolean hasProcessGroupContents(final VersionedProcessGroup group) {
+        if (group == null) {
+            return false;
+        }
+
+        return !group.getConnections().isEmpty()
+                || !group.getProcessors().isEmpty()
+                || !group.getControllerServices().isEmpty()
+                || !group.getFunnels().isEmpty()
+                || !group.getInputPorts().isEmpty()
+                || !group.getLabels().isEmpty()
+                || !group.getOutputPorts().isEmpty()
+                || !group.getProcessGroups().isEmpty()
+                || !group.getRemoteProcessGroups().isEmpty();
+    }
+
 
     private void compareFlowCoordinates(final VersionedProcessGroup groupA, final VersionedProcessGroup groupB, final Set<FlowDifference> differences) {
         final VersionedFlowCoordinates coordinatesA = groupA.getVersionedFlowCoordinates();

nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/TestStandardFlowComparator.java

@@ -23,6 +23,7 @@
 import org.apache.nifi.flow.VersionedAsset;
 import org.apache.nifi.flow.VersionedComponent;
 import org.apache.nifi.flow.VersionedControllerService;
+import org.apache.nifi.flow.VersionedFlowCoordinates;
 import org.apache.nifi.flow.VersionedParameter;
 import org.apache.nifi.flow.VersionedParameterContext;
 import org.apache.nifi.flow.VersionedProcessGroup;
@@ -285,6 +286,63 @@ public void testDeepStrategyWithChildPGs() {
                         && difference.getComponentA().getComponentType() == ComponentType.CONTROLLER_SERVICE));
     }
 
+    @Test
+    public void testScheduledStateChangeDetectedForProcessorInNestedVersionedGroup() {
+        final String rootPgIdentifier = "rootPG";
+        final String nestedPgIdentifier = "nestedPG";
+        final String procIdentifier = "processorZ";
+        final VersionedProcessGroup registryRoot = new VersionedProcessGroup();
+        registryRoot.setIdentifier(rootPgIdentifier);
+
+        final VersionedProcessGroup localRoot = new VersionedProcessGroup();
+        localRoot.setIdentifier(rootPgIdentifier);
+
+        final VersionedProcessGroup registryNested = new VersionedProcessGroup();
+        registryNested.setIdentifier(nestedPgIdentifier);
+        registryNested.setVersionedFlowCoordinates(createVersionedFlowCoordinates());
+        registryRoot.getProcessGroups().add(registryNested);
+
+        final VersionedProcessGroup localNested = new VersionedProcessGroup();
+        localNested.setIdentifier(nestedPgIdentifier);
+        localNested.setVersionedFlowCoordinates(createVersionedFlowCoordinates());
+        localRoot.getProcessGroups().add(localNested);
+
+        final VersionedProcessor registryProcessor = new VersionedProcessor();
+        registryProcessor.setIdentifier(procIdentifier);
+        registryProcessor.setScheduledState(ScheduledState.ENABLED);
+        registryProcessor.setProperties(Collections.emptyMap());
+        registryProcessor.setPropertyDescriptors(Collections.emptyMap());
+        registryNested.getProcessors().add(registryProcessor);
+
+        final VersionedProcessor localProcessor = new VersionedProcessor();
+        localProcessor.setIdentifier(procIdentifier);
+        localProcessor.setScheduledState(ScheduledState.DISABLED);
+        localProcessor.setProperties(Collections.emptyMap());
+        localProcessor.setPropertyDescriptors(Collections.emptyMap());
+        localNested.getProcessors().add(localProcessor);
+
+        final ComparableDataFlow registryFlow = new StandardComparableDataFlow("registry", registryRoot);
+        final ComparableDataFlow localFlow = new StandardComparableDataFlow("local", localRoot);
+
+        final StandardFlowComparator testComparator = new StandardFlowComparator(
+                registryFlow,
+                localFlow,
+                Collections.emptySet(),
+                new StaticDifferenceDescriptor(),
+                Function.identity(),
+                VersionedComponent::getIdentifier,
+                FlowComparatorVersionedStrategy.SHALLOW);
+
+        final Set<FlowDifference> differences = testComparator.compare().getDifferences();
+
+        final boolean scheduledStateDiffFound = differences.stream()
+                .anyMatch(diff -> diff.getDifferenceType() == DifferenceType.SCHEDULED_STATE_CHANGED
+                        && diff.getComponentB() != null
+                        && procIdentifier.equals(diff.getComponentB().getIdentifier()));
+
+        assertTrue(scheduledStateDiffFound, "Expected scheduled state change for processor inside nested process group to be detected");
+    }
+
     private VersionedParameter createParameter(final String name, final String value, final boolean sensitive) {
         return createParameter(name, value, sensitive, null);
     }
@@ -304,4 +362,13 @@ private VersionedAsset createAsset(final String id, final String name) {
         asset.setName(name);
         return asset;
     }
+
+    private VersionedFlowCoordinates createVersionedFlowCoordinates() {
+        final VersionedFlowCoordinates coordinates = new VersionedFlowCoordinates();
+        coordinates.setRegistryId("registry");
+        coordinates.setBucketId("bucketId");
+        coordinates.setFlowId("flowId");
+        coordinates.setVersion("1");
+        return coordinates;
+    }
 }