diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java index a43aca79a8ba..4f2112c6bfd1 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java @@ -344,6 +344,14 @@ public void synchronize(final ProcessGroup group, final VersionedExternalFlow ve continue; } + // When updating from version control, preserve a local rename of a public port (an input/output port that allows remote + // access) instead of reverting it to the registry-defined name. Without this, the name change is treated as an update and the user's + // local name is overwritten. Only the version-control update path opts in via preservePublicPortNames; cluster reconnection and startup + // leave it false so the node still adopts the incoming flow's port names. + if (syncOptions.isPreservePublicPortNames() && FlowDifferenceFilters.isPublicPortNameChange(diff)) { + continue; + } + // If this update adds a new Controller Service, then we need to check if the service already exists at a higher level // and if so compare our VersionedControllerService to the existing service. if (diff.getDifferenceType() == DifferenceType.COMPONENT_ADDED) { @@ -1042,7 +1050,10 @@ private void synchronizeInputPorts(final ProcessGroup group, final VersionedProc LOG.info("Added {} to {}", added, group); } else if (updatedVersionedComponentIds.contains(proposedPort.getIdentifier())) { final String temporaryName = generateTemporaryPortName(proposedPort); - proposedPortFinalNames.put(port, proposedPort.getName()); + // When the port is updated for any reason, preserve the local name of a public port instead of overwriting it with the + // registry-defined name (the port may be in the update set because of another difference such as a comment change). + final String finalName = syncOptions.isPreservePublicPortNames() && port instanceof PublicPort ? port.getName() : proposedPort.getName(); + proposedPortFinalNames.put(port, finalName); updatePort(port, proposedPort, temporaryName); LOG.info("Updated {}", port); } else { @@ -1064,7 +1075,10 @@ private void synchronizeOutputPorts(final ProcessGroup group, final VersionedPro LOG.info("Added {} to {}", added, group); } else if (updatedVersionedComponentIds.contains(proposedPort.getIdentifier())) { final String temporaryName = generateTemporaryPortName(proposedPort); - proposedPortFinalNames.put(port, proposedPort.getName()); + // When the port is updated for any reason, preserve the local name of a public port instead of overwriting it with the + // registry-defined name (the port may be in the update set because of another difference such as a comment change). + final String finalName = syncOptions.isPreservePublicPortNames() && port instanceof PublicPort ? port.getName() : proposedPort.getName(); + proposedPortFinalNames.put(port, finalName); updatePort(port, proposedPort, temporaryName); LOG.info("Updated {}", port); } else { diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java index dc46a0997037..287ea7c6b309 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java @@ -3959,7 +3959,10 @@ public void updateFlow(final VersionedExternalFlow proposedSnapshot, final Strin .ignoreLocalModifications(!verifyNotDirty) .updateDescendantVersionedFlows(updateDescendantVersionedFlows) .updateGroupSettings(updateSettings) - .updateRpgUrls(false); + .updateRpgUrls(false) + // A user may rename a public port locally (e.g. to resolve a name collision when reusing a versioned group). That local name must + // survive a version-control update rather than being reverted to the name stored in the registry. + .preservePublicPortNames(true); // Connectors should not have encrypted values copied from versioned flow. However we do need to decrypt parameter references. if (getConnectorIdentifier().isPresent()) { flowSynchronizationBuilder.propertyDecryptor(value -> value); diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java index 9ad6c27646e7..0c2f28f7aed0 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java @@ -78,6 +78,7 @@ import org.apache.nifi.processor.ProcessContext; import org.apache.nifi.processor.Relationship; import org.apache.nifi.registry.flow.mapping.FlowMappingOptions; +import org.apache.nifi.remote.PublicPort; import org.apache.nifi.reporting.InitializationException; import org.apache.nifi.scheduling.ExecutionNode; import org.apache.nifi.scheduling.SchedulingStrategy; @@ -935,6 +936,56 @@ public void testStoppedPortNotRestarted() throws FlowSynchronizationException, I verify(inputPort).setName("Input"); } + @Test + public void testPublicInputPortLocalNamePreservedOnVersionControlUpdate() { + // With preservePublicPortNames on (the version-control update path), a public input port whose only difference + // from the registry flow is a local rename must keep its local name; the port is not even flagged for update. + final ProcessGroup processGroup = createMockProcessGroup(); + final PublicPort publicPort = createMappablePublicInputPort(processGroup, "port-vid", "Local Renamed", null); + when(processGroup.getInputPorts()).thenReturn(Collections.singleton(publicPort)); + when(flowManager.getPublicInputPort(anyString())).thenReturn(Optional.empty()); + + final VersionedExternalFlow externalFlow = singlePublicInputPortFlow(processGroup, "port-vid", "Registry Name", null); + + assertDoesNotThrow(() -> synchronizer.synchronize(processGroup, externalFlow, preservePublicPortNamesOptions(true))); + + verify(publicPort, never()).setName(anyString()); + } + + @Test + public void testPublicInputPortNameAdoptedWhenNotPreserving() { + // With preservePublicPortNames off (cluster reconnection / startup), the incoming flow's port name must still be adopted. + final ProcessGroup processGroup = createMockProcessGroup(); + final PublicPort publicPort = createMappablePublicInputPort(processGroup, "port-vid", "Local Renamed", null); + when(processGroup.getInputPorts()).thenReturn(Collections.singleton(publicPort)); + when(flowManager.getPublicInputPort(anyString())).thenReturn(Optional.empty()); + + final VersionedExternalFlow externalFlow = singlePublicInputPortFlow(processGroup, "port-vid", "Registry Name", null); + + assertDoesNotThrow(() -> synchronizer.synchronize(processGroup, externalFlow, preservePublicPortNamesOptions(false))); + + verify(publicPort).setName("Registry Name"); + } + + @Test + public void testPublicInputPortLocalNamePreservedWhenPortUpdatedForOtherReasons() { + // Multi-change case: the port lands in the update set because of another difference (comments), but with the + // flag on the local name must still be preserved even though the port is updated. + final ProcessGroup processGroup = createMockProcessGroup(); + final PublicPort publicPort = createMappablePublicInputPort(processGroup, "port-vid", "Local Renamed", "old comments"); + when(processGroup.getInputPorts()).thenReturn(Collections.singleton(publicPort)); + when(flowManager.getPublicInputPort(anyString())).thenReturn(Optional.empty()); + + final VersionedExternalFlow externalFlow = singlePublicInputPortFlow(processGroup, "port-vid", "Registry Name", "new comments"); + + assertDoesNotThrow(() -> synchronizer.synchronize(processGroup, externalFlow, preservePublicPortNamesOptions(true))); + + // The comment change is applied, but the local name wins over the registry name. + verify(publicPort).setComments("new comments"); + verify(publicPort).setName("Local Renamed"); + verify(publicPort, never()).setName("Registry Name"); + } + @Test public void testRemoveOutputPortFailsIfIncomingConnection() { createMockConnection(processorA, outputPort, group); @@ -1674,6 +1725,57 @@ private VersionedPort createMinimalVersionedPort(final ComponentType componentTy return versionedPort; } + private FlowSynchronizationOptions preservePublicPortNamesOptions(final boolean preserve) { + return new FlowSynchronizationOptions.Builder() + .componentIdGenerator(componentIdGenerator) + .componentComparisonIdLookup(VersionedComponent::getIdentifier) + .componentScheduler(componentScheduler) + .scheduledStateChangeListener(scheduledStateChangeListener) + .preservePublicPortNames(preserve) + .build(); + } + + private PublicPort createMappablePublicInputPort(final ProcessGroup processGroup, final String versionedId, final String localName, final String comments) { + final String groupId = processGroup.getIdentifier(); + final PublicPort port = Mockito.mock(PublicPort.class); + when(port.getIdentifier()).thenReturn(UUID.randomUUID().toString()); + when(port.getProcessGroupIdentifier()).thenReturn(groupId); + when(port.getVersionedComponentId()).thenReturn(Optional.of(versionedId)); + when(port.getName()).thenReturn(localName); + when(port.getComments()).thenReturn(comments); + when(port.getMaxConcurrentTasks()).thenReturn(1); + when(port.getPosition()).thenReturn(new org.apache.nifi.connectable.Position(0, 0)); + when(port.getConnectableType()).thenReturn(ConnectableType.INPUT_PORT); + when(port.getScheduledState()).thenReturn(org.apache.nifi.controller.ScheduledState.STOPPED); + when(port.isRunning()).thenReturn(false); + when(port.getProcessGroup()).thenReturn(processGroup); + return port; + } + + // Builds a proposed flow whose only public input port matches the live port by versioned id, differing by the given name + // (and optionally comments). ENABLED scheduled state and the other fields match what a STOPPED live port maps to, so the + // only differences are the ones under test. + private VersionedExternalFlow singlePublicInputPortFlow(final ProcessGroup processGroup, final String versionedId, final String name, final String comments) { + final VersionedPort proposedPort = new VersionedPort(); + proposedPort.setIdentifier(versionedId); + proposedPort.setInstanceIdentifier(versionedId); + proposedPort.setName(name); + proposedPort.setComments(comments); + proposedPort.setScheduledState(ScheduledState.ENABLED); + proposedPort.setComponentType(ComponentType.INPUT_PORT); + proposedPort.setPosition(new Position(0D, 0D)); + proposedPort.setConcurrentlySchedulableTaskCount(1); + proposedPort.setAllowRemoteAccess(Boolean.TRUE); + + final VersionedProcessGroup versionedGroup = new VersionedProcessGroup(); + versionedGroup.setIdentifier(processGroup.getIdentifier()); + versionedGroup.setInputPorts(Set.of(proposedPort)); + + final VersionedExternalFlow externalFlow = new VersionedExternalFlow(); + externalFlow.setFlowContents(versionedGroup); + return externalFlow; + } + private void assertSensitivePropertyDecrypted(final ComponentNode componentNode) { verify(componentNode).setProperties(propertiesCaptor.capture(), eq(true), eq(Collections.emptySet())); diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/util/TestFlowDifferenceFilters.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/util/TestFlowDifferenceFilters.java index dfb672b0c05d..81c45668eef0 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/util/TestFlowDifferenceFilters.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/util/TestFlowDifferenceFilters.java @@ -949,6 +949,28 @@ public void testIsComponentUpdateRequiredForLocalScheduleStateChange() { assertTrue(FlowDifferenceFilters.isComponentUpdateRequired(scheduledStateDiff, null, flowManager)); } + @Test + public void testIsComponentUpdateRequiredForPublicPortNameChange() { + // A public-port name change must still be reported as "update required" by the shared filter. The preservation of a + // local public-port name is scoped to the synchronizer (gated by FlowSynchronizationOptions.preservePublicPortNames) and to the + // affected-components calculation (which applies FILTER_PUBLIC_PORT_NAME_CHANGES), so the core semantics of this shared, static + // method must remain unchanged - otherwise cluster reconnection and startup would stop preserving the incoming flow's port names. + final FlowManager flowManager = Mockito.mock(FlowManager.class); + + final VersionedPort portA = new VersionedPort(); + portA.setAllowRemoteAccess(Boolean.TRUE); + final VersionedPort portB = new VersionedPort(); + portB.setAllowRemoteAccess(Boolean.TRUE); + + final StandardFlowDifference nameChange = new StandardFlowDifference( + DifferenceType.NAME_CHANGED, portA, portB, "Original Name", "Renamed", ""); + + assertTrue(FlowDifferenceFilters.isComponentUpdateRequired(nameChange, null, flowManager)); + + // The dedicated predicate, on the other hand, excludes the public-port name change so opt-in callers can preserve the local name. + assertFalse(FlowDifferenceFilters.FILTER_PUBLIC_PORT_NAME_CHANGES.test(nameChange)); + } + @DynamicProperty(name = "Dynamic Property", value = "Value", description = "Allows dynamic properties") private static class DynamicAnnotationProcessor extends AbstractProcessor { @Override diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/FlowSynchronizationOptions.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/FlowSynchronizationOptions.java index b857f07f053d..aaa19cf30462 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/FlowSynchronizationOptions.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/FlowSynchronizationOptions.java @@ -31,6 +31,7 @@ public class FlowSynchronizationOptions { private final boolean updateSettings; private final boolean updateDescendantVersionedFlows; private final boolean updateRpgUrls; + private final boolean preservePublicPortNames; private final Duration componentStopTimeout; private final ComponentStopTimeoutAction timeoutAction; private final ScheduledStateChangeListener scheduledStateChangeListener; @@ -45,6 +46,7 @@ private FlowSynchronizationOptions(final Builder builder) { this.updateSettings = builder.updateSettings; this.updateDescendantVersionedFlows = builder.updateDescendantVersionedFlows; this.updateRpgUrls = builder.updateRpgUrls; + this.preservePublicPortNames = builder.preservePublicPortNames; this.componentStopTimeout = builder.componentStopTimeout; this.timeoutAction = builder.timeoutAction; this.scheduledStateChangeListener = builder.scheduledStateChangeListener; @@ -79,6 +81,10 @@ public boolean isUpdateRpgUrls() { return updateRpgUrls; } + public boolean isPreservePublicPortNames() { + return preservePublicPortNames; + } + public PropertyDecryptor getPropertyDecryptor() { return propertyDecryptor; } @@ -107,6 +113,7 @@ public static class Builder { private boolean updateSettings = true; private boolean updateDescendantVersionedFlows = true; private boolean updateRpgUrls = false; + private boolean preservePublicPortNames = false; private ScheduledStateChangeListener scheduledStateChangeListener; private PropertyDecryptor propertyDecryptor = value -> value; private Duration componentStopTimeout = Duration.ofSeconds(30); @@ -188,6 +195,20 @@ public Builder updateRpgUrls(final boolean updateRpgUrls) { return this; } + /** + * Specifies whether the local name of a public port (an input/output port that allows remote access) should be preserved when synchronizing, + * rather than being overwritten with the name from the proposed flow. This is used for registry version-control updates, where a user may have + * renamed a public port locally to avoid a name collision, and that local name must survive the update. It should remain false for cluster + * reconnection and startup flow inheritance, where the node must adopt the incoming flow's port names verbatim. + * + * @param preservePublicPortNames whether to preserve local public-port names + * @return the builder + */ + public Builder preservePublicPortNames(final boolean preservePublicPortNames) { + this.preservePublicPortNames = preservePublicPortNames; + return this; + } + /** * Specifies the decryptor to use for sensitive properties * @@ -265,6 +286,7 @@ public static Builder from(final FlowSynchronizationOptions options) { builder.updateSettings = options.isUpdateSettings(); builder.updateDescendantVersionedFlows = options.isUpdateDescendantVersionedFlows(); builder.updateRpgUrls = options.isUpdateRpgUrls(); + builder.preservePublicPortNames = options.isPreservePublicPortNames(); builder.propertyDecryptor = options.getPropertyDecryptor(); builder.componentStopTimeout = options.getComponentStopTimeout(); builder.timeoutAction = options.getComponentStopTimeoutAction(); diff --git a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java index 4a50049e5dec..6386ee0c6f11 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java +++ b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java @@ -6847,6 +6847,10 @@ public Set getComponentsAffectedByFlowUpdate(final Stri .filter(FlowDifferenceFilters.FILTER_ADDED_REMOVED_REMOTE_PORTS) .filter(difference -> difference.getComponentA() != null) // a difference that would not affect a local component .filter(diff -> FlowDifferenceFilters.isComponentUpdateRequired(diff, proposedFlow.getContents(), flowManager)) + // A local rename of a public port is preserved during a version-control update (it is not overwritten with the + // registry name), so the port must not be reported as affected/stopped for that name change. Applied unconditionally here because + // this affected-components calculation serves only the version-control update path. + .filter(FlowDifferenceFilters.FILTER_PUBLIC_PORT_NAME_CHANGES) .filter(diff -> !FlowDifferenceFilters.isLocalScheduleStateChange(diff)) .map(difference -> { final VersionedComponent localComponent = difference.getComponentA();