This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new c7c745e8c8d NIFI-16064 Preserve local Public Port Names on updates
from Version Control (#11384)
c7c745e8c8d is described below
commit c7c745e8c8dcc7518b9d03720c85879cf29281be
Author: Pierre Villard <[email protected]>
AuthorDate: Wed Jul 22 22:12:30 2026 +0200
NIFI-16064 Preserve local Public Port Names on updates from Version Control
(#11384)
Signed-off-by: David Handermann <[email protected]>
---
.../StandardVersionedComponentSynchronizer.java | 18 +++-
.../apache/nifi/groups/StandardProcessGroup.java | 5 +-
...StandardVersionedComponentSynchronizerTest.java | 102 +++++++++++++++++++++
.../nifi/util/TestFlowDifferenceFilters.java | 22 +++++
.../nifi/groups/FlowSynchronizationOptions.java | 22 +++++
.../apache/nifi/web/StandardNiFiServiceFacade.java | 4 +
6 files changed, 170 insertions(+), 3 deletions(-)
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 06442bd4757..7315d8d441d 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
@@ -343,6 +343,14 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
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) {
@@ -1051,7 +1059,10 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
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 {
@@ -1073,7 +1084,10 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
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 6ef66c0ea0d..1aa7d4c5959 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 final class StandardProcessGroup implements
ProcessGroup {
.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 9ad6c27646e..0c2f28f7aed 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.parameter.StandardParameterContextManager;
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 class StandardVersionedComponentSynchronizerTest {
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 @@ public class StandardVersionedComponentSynchronizerTest {
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 dfb672b0c05..81c45668eef 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 class TestFlowDifferenceFilters {
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 d2e8624b04d..a9e1014d45f 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 @@ public class FlowSynchronizationOptions {
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 class FlowSynchronizationOptions {
return updateRpgUrls;
}
+ public boolean isPreservePublicPortNames() {
+ return preservePublicPortNames;
+ }
+
public PropertyDecryptor getPropertyDecryptor() {
return propertyDecryptor;
}
@@ -107,6 +113,7 @@ public class FlowSynchronizationOptions {
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 class FlowSynchronizationOptions {
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 class FlowSynchronizationOptions {
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 5ae791ae3b3..aca6949c908 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
@@ -6919,6 +6919,10 @@ public class StandardNiFiServiceFacade implements
NiFiServiceFacade {
.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();
