Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
exceptionfactory merged 1 commit into from
Oct 2, 2025
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 18 additions & 2 deletions ...ry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardFlowComparator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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();
Expand Down
67 changes: 67 additions & 0 deletions ...low-diff/src/test/java/org/apache/nifi/registry/flow/diff/TestStandardFlowComparator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Comment thread
pvillard31 marked this conversation as resolved.

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);
}
Expand All @@ -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;
}
}