diff --git a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardFlowComparator.java b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardFlowComparator.java index 6c3145670748..434142f18eec 100644 --- a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardFlowComparator.java +++ b/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 differences) { final VersionedFlowCoordinates coordinatesA = groupA.getVersionedFlowCoordinates(); diff --git a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/TestStandardFlowComparator.java b/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/TestStandardFlowComparator.java index 667f6822e053..f5d19a0a2e95 100644 --- a/nifi-registry/nifi-registry-core/nifi-registry-flow-diff/src/test/java/org/apache/nifi/registry/flow/diff/TestStandardFlowComparator.java +++ b/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 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; + } }