NIFI-16064 - Preserve local public port names on version control update

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 {

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);

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()));
 

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

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();

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<AffectedComponentEntity> 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();