From 049eef2c0445b0a4221ca7ea71555842d4ee3ae0 Mon Sep 17 00:00:00 2001 From: Pierre Villard Date: Tue, 7 Jul 2026 10:31:46 +0200 Subject: [PATCH] NIFI-16082 - Connection update is incorrectly rejected when the current destination is running --- .../nifi/connectable/StandardConnection.java | 35 +++-- .../connectable/StandardConnectionTest.java | 129 +++++++++++++++++ .../apache/nifi/connectable/Connection.java | 10 ++ .../web/dao/impl/StandardConnectionDAO.java | 57 +++++--- .../dao/impl/StandardConnectionDAOTest.java | 131 ++++++++++++++++++ 5 files changed, 329 insertions(+), 33 deletions(-) create mode 100644 nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/connectable/StandardConnectionTest.java diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/connectable/StandardConnection.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/connectable/StandardConnection.java index 4727d60404e1..39343b48436d 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/connectable/StandardConnection.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/connectable/StandardConnection.java @@ -292,6 +292,27 @@ public void setDestination(final Connectable newDestination) { return; } + verifyCanUpdateDestination(); + + if (newDestination instanceof Funnel && newDestination.equals(source)) { + throw new IllegalStateException("Funnels do not support self-looping connections."); + } + + try { + previousDestination.removeConnection(this); + this.destination.set(newDestination); + getSource().updateConnection(this); + newDestination.addConnection(this); + } catch (final RuntimeException e) { + this.destination.set(previousDestination); + throw e; + } + } + + @Override + public void verifyCanUpdateDestination() { + final Connectable previousDestination = destination.get(); + // Allow destination changes when the current destination is a Funnel, a LocalPort, or a // RemoteGroupPort. Funnels and LocalPorts cannot be stopped/started so they are exempt. // RemoteGroupPort represents an S2S ingress point: re-routing its incoming connections @@ -310,20 +331,6 @@ public void setDestination(final Connectable newDestination) { if (getFlowFileQueue().isUnacknowledgedFlowFile()) { throw new IllegalStateException("Cannot change destination of Connection because FlowFiles from this Connection are currently held by " + previousDestination); } - - if (newDestination instanceof Funnel && newDestination.equals(source)) { - throw new IllegalStateException("Funnels do not support self-looping connections."); - } - - try { - previousDestination.removeConnection(this); - this.destination.set(newDestination); - getSource().updateConnection(this); - newDestination.addConnection(this); - } catch (final RuntimeException e) { - this.destination.set(previousDestination); - throw e; - } } @Override diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/connectable/StandardConnectionTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/connectable/StandardConnectionTest.java new file mode 100644 index 000000000000..e2f9d334973f --- /dev/null +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/connectable/StandardConnectionTest.java @@ -0,0 +1,129 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.connectable; + +import org.apache.nifi.controller.ProcessScheduler; +import org.apache.nifi.controller.queue.FlowFileQueue; +import org.apache.nifi.controller.queue.FlowFileQueueFactory; +import org.apache.nifi.groups.ProcessGroup; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.remote.RemoteGroupPort; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class StandardConnectionTest { + + private FlowFileQueue queue; + + private StandardConnection connectionWithDestination(final Connectable source, final Connectable destination) { + queue = mock(FlowFileQueue.class); + final FlowFileQueueFactory queueFactory = mock(FlowFileQueueFactory.class); + when(queueFactory.createFlowFileQueue(any(), any(), any())).thenReturn(queue); + + return new StandardConnection.Builder(mock(ProcessScheduler.class)) + .id("connection-1") + .source(source) + .destination(destination) + .processGroup(mock(ProcessGroup.class)) + .flowFileQueueFactory(queueFactory) + .relationships(List.of(Relationship.ANONYMOUS)) + .build(); + } + + private StandardConnection connectionWithDestination(final Connectable destination) { + return connectionWithDestination(mock(Connectable.class), destination); + } + + @Test + void testVerifyCanUpdateDestinationThrowsWhenCurrentDestinationRunning() { + final Connectable runningProcessor = mock(Connectable.class); + when(runningProcessor.isRunning()).thenReturn(true); + final StandardConnection connection = connectionWithDestination(runningProcessor); + + assertThrows(IllegalStateException.class, connection::verifyCanUpdateDestination); + } + + @Test + void testVerifyCanUpdateDestinationAllowsRunningFunnel() { + final Funnel runningFunnel = mock(Funnel.class); + when(runningFunnel.isRunning()).thenReturn(true); + final StandardConnection connection = connectionWithDestination(runningFunnel); + + assertDoesNotThrow(connection::verifyCanUpdateDestination); + } + + @Test + void testVerifyCanUpdateDestinationAllowsRunningLocalPort() { + final LocalPort runningLocalPort = mock(LocalPort.class); + when(runningLocalPort.isRunning()).thenReturn(true); + final StandardConnection connection = connectionWithDestination(runningLocalPort); + + assertDoesNotThrow(connection::verifyCanUpdateDestination); + } + + @Test + void testVerifyCanUpdateDestinationAllowsRunningRemoteGroupPort() { + final RemoteGroupPort runningRemotePort = mock(RemoteGroupPort.class); + when(runningRemotePort.isRunning()).thenReturn(true); + final StandardConnection connection = connectionWithDestination(runningRemotePort); + + assertDoesNotThrow(connection::verifyCanUpdateDestination); + } + + @Test + void testVerifyCanUpdateDestinationThrowsWhenFlowFilesHeld() { + final Funnel stoppedFunnel = mock(Funnel.class); + final StandardConnection connection = connectionWithDestination(stoppedFunnel); + when(queue.isUnacknowledgedFlowFile()).thenReturn(true); + + assertThrows(IllegalStateException.class, connection::verifyCanUpdateDestination); + } + + @Test + void testVerifyCanUpdateDestinationAllowsStoppedDestinationWithNoUnacknowledgedFlowFiles() { + final Connectable stoppedProcessor = mock(Connectable.class); + final StandardConnection connection = connectionWithDestination(stoppedProcessor); + when(queue.isUnacknowledgedFlowFile()).thenReturn(false); + + assertDoesNotThrow(connection::verifyCanUpdateDestination); + } + + @Test + void testSetDestinationNoOpWhenDestinationUnchanged() { + final Connectable runningProcessor = mock(Connectable.class); + when(runningProcessor.isRunning()).thenReturn(true); + final StandardConnection connection = connectionWithDestination(runningProcessor); + + assertDoesNotThrow(() -> connection.setDestination(runningProcessor)); + } + + @Test + void testSetDestinationRejectsSelfLoopingFunnel() { + final Funnel funnel = mock(Funnel.class); + final Connectable currentDestination = mock(Connectable.class); + final StandardConnection connection = connectionWithDestination(funnel, currentDestination); + + assertThrows(IllegalStateException.class, () -> connection.setDestination(funnel)); + } +} diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/connectable/Connection.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/connectable/Connection.java index 423f52d4ceca..434e130db07a 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/connectable/Connection.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/connectable/Connection.java @@ -68,6 +68,16 @@ public interface Connection extends Authorizable, VersionedComponent { void setDestination(final Connectable newDestination); + /** + * Verifies that this Connection's destination may be changed, based solely on the current (existing) destination + * and the FlowFiles the Connection is holding. This applies the same guards as {@link #setDestination(Connectable)} + * so that a pre-check (e.g. the two-phase cluster verify) rejects exactly what the subsequent mutation would reject. + * + * @throws IllegalStateException if the current destination is running and is not exempt, or FlowFiles from this + * Connection are currently held by the destination + */ + void verifyCanUpdateDestination() throws IllegalStateException; + void setProcessGroup(ProcessGroup processGroup); ProcessGroup getProcessGroup(); diff --git a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardConnectionDAO.java b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardConnectionDAO.java index 8b949cb6da9c..4354437a684a 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardConnectionDAO.java +++ b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardConnectionDAO.java @@ -519,12 +519,16 @@ private void verifyUpdate(final Connection connection, final ConnectionDTO conne throw new ValidationException(validationErrors); } - // If destination is changing, ensure that current destination is not running. This check is done here, rather than - // in the Connection object itself because the Connection object itself does not know which updates are to occur and - // we don't want to prevent updating things like the connection name or backpressure just because the destination is running - final Connectable destination = connection.getDestination(); - if (destination != null && destination.isRunning() && destination.getConnectableType() != ConnectableType.FUNNEL && destination.getConnectableType() != ConnectableType.INPUT_PORT) { - throw new ValidationException(Collections.singletonList("Cannot change the destination of connection because the current destination is running")); + // If the destination is changing, ensure the current destination is in a state that allows it to be changed. + // The check lives on the Connection (shared with setDestination) so the verify phase rejects exactly what the + // subsequent mutation would reject. It is applied only for an actual destination change so that other updates + // (name, backpressure, etc.) are not blocked merely because the current destination is running. + if (isDestinationChanging(connection, connectionDTO.getDestination())) { + try { + connection.verifyCanUpdateDestination(); + } catch (final IllegalStateException e) { + throw new ValidationException(Collections.singletonList(e.getMessage())); + } } // verify that this connection supports modification @@ -532,6 +536,32 @@ private void verifyUpdate(final Connection connection, final ConnectionDTO conne } } + /** + * Determines whether the proposed destination represents an actual change from the Connection's current destination. + * Mirrors the change-detection used by {@link #updateConnection(ConnectionDTO)} so the verify and commit phases agree + * on what constitutes a destination change — including a remote input port re-pointed to a different remote process group. + */ + boolean isDestinationChanging(final Connection connection, final ConnectableDTO proposedDestination) { + if (proposedDestination == null) { + return false; + } + + final Connectable currentDestination = connection.getDestination(); + if (!proposedDestination.getId().equals(currentDestination.getIdentifier())) { + return true; + } + + // Same destination id: for a remote input port, a different remote process group id is also a change. + if (ConnectableType.REMOTE_INPUT_PORT.name().equals(proposedDestination.getType()) + && currentDestination.getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { + final RemoteGroupPort remotePort = (RemoteGroupPort) currentDestination; + return proposedDestination.getGroupId() != null + && !proposedDestination.getGroupId().equals(remotePort.getRemoteProcessGroup().getIdentifier()); + } + + return false; + } + @Override public Connection updateConnection(final ConnectionDTO connectionDTO) { final Connection connection = locateConnection(connectionDTO.getId()); @@ -571,8 +601,6 @@ public Connection updateConnection(final ConnectionDTO connectionDTO) { // determine if the destination changed final ConnectableDTO proposedDestination = connectionDTO.getDestination(); if (proposedDestination != null) { - final Connectable currentDestination = connection.getDestination(); - // handle remote input port differently if (ConnectableType.REMOTE_INPUT_PORT.name().equals(proposedDestination.getType())) { // the group id must be specified @@ -580,17 +608,8 @@ public Connection updateConnection(final ConnectionDTO connectionDTO) { throw new IllegalArgumentException("When the destination is a remote input port its group id is required."); } - // if the current destination is a remote input port - boolean isDifferentRemoteProcessGroup = false; - if (currentDestination.getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { - RemoteGroupPort remotePort = (RemoteGroupPort) currentDestination; - if (!proposedDestination.getGroupId().equals(remotePort.getRemoteProcessGroup().getIdentifier())) { - isDifferentRemoteProcessGroup = true; - } - } - // if the destination is changing or the previous destination was a different remote process group - if (!proposedDestination.getId().equals(currentDestination.getIdentifier()) || isDifferentRemoteProcessGroup) { + if (isDestinationChanging(connection, proposedDestination)) { final ProcessGroup destinationParentGroup = locateProcessGroup(flowController, group.getIdentifier()); final RemoteProcessGroup remoteProcessGroup = destinationParentGroup.getRemoteProcessGroup(proposedDestination.getGroupId()); @@ -615,7 +634,7 @@ public Connection updateConnection(final ConnectionDTO connectionDTO) { } } else { // if there is a different destination id - if (!proposedDestination.getId().equals(currentDestination.getIdentifier())) { + if (isDestinationChanging(connection, proposedDestination)) { // if the destination connectable's group id has not been set, its inferred to be the current group if (proposedDestination.getGroupId() == null) { proposedDestination.setGroupId(group.getIdentifier()); diff --git a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/dao/impl/StandardConnectionDAOTest.java b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/dao/impl/StandardConnectionDAOTest.java index 20df04a91601..abe9dea4d9bc 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/dao/impl/StandardConnectionDAOTest.java +++ b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/dao/impl/StandardConnectionDAOTest.java @@ -21,11 +21,18 @@ import org.apache.nifi.components.connector.ConnectorState; import org.apache.nifi.components.connector.ConnectorSyncMode; import org.apache.nifi.components.connector.FrameworkFlowContext; +import org.apache.nifi.connectable.Connectable; +import org.apache.nifi.connectable.ConnectableType; import org.apache.nifi.connectable.Connection; import org.apache.nifi.controller.FlowController; +import org.apache.nifi.controller.exception.ValidationException; import org.apache.nifi.controller.flow.FlowManager; import org.apache.nifi.groups.ProcessGroup; +import org.apache.nifi.groups.RemoteProcessGroup; +import org.apache.nifi.remote.RemoteGroupPort; import org.apache.nifi.web.ResourceNotFoundException; +import org.apache.nifi.web.api.dto.ConnectableDTO; +import org.apache.nifi.web.api.dto.ConnectionDTO; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -37,10 +44,14 @@ import java.util.List; import java.util.Optional; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @ExtendWith(MockitoExtension.class) @@ -199,4 +210,124 @@ void testGetConnectionWithMultipleConnectors() { assertEquals(connectionInSecondConnector, result); } + + private ConnectionDTO connectionDtoWithNameOnly() { + final ConnectionDTO dto = new ConnectionDTO(); + dto.setId(ROOT_CONNECTION_ID); + dto.setName("renamed-connection"); + return dto; + } + + private ConnectionDTO connectionDtoChangingDestination(final String newDestinationId) { + final ConnectionDTO dto = new ConnectionDTO(); + dto.setId(ROOT_CONNECTION_ID); + final ConnectableDTO newDestination = new ConnectableDTO(); + newDestination.setId(newDestinationId); + newDestination.setType(ConnectableType.PROCESSOR.name()); + dto.setDestination(newDestination); + return dto; + } + + private void stubRootConnectionDestination(final String destinationId) { + final ProcessGroup group = mock(ProcessGroup.class); + when(group.getIdentifier()).thenReturn("group-id"); + when(rootConnection.getProcessGroup()).thenReturn(group); + + final Connectable currentDestination = mock(Connectable.class); + when(currentDestination.getIdentifier()).thenReturn(destinationId); + when(currentDestination.isRunning()).thenReturn(true); + when(currentDestination.getConnectableType()).thenReturn(ConnectableType.PROCESSOR); + when(rootConnection.getDestination()).thenReturn(currentDestination); + } + + @Test + void testVerifyUpdateDoesNotCheckDestinationForNonDestinationEdit() { + stubRootConnectionDestination("current-destination-id"); + + assertDoesNotThrow(() -> connectionDAO.verifyUpdate(connectionDtoWithNameOnly())); + verify(rootConnection, never()).verifyCanUpdateDestination(); + } + + @Test + void testVerifyUpdateWrapsIllegalStateFromDestinationGuardAsValidationException() { + stubRootConnectionDestination("current-destination-id"); + final String guardMessage = "Cannot change destination of Connection because the current destination ([proc]) is running"; + org.mockito.Mockito.doThrow(new IllegalStateException(guardMessage)) + .when(rootConnection).verifyCanUpdateDestination(); + + final ValidationException thrown = assertThrows(ValidationException.class, + () -> connectionDAO.verifyUpdate(connectionDtoChangingDestination("new-destination-id"))); + assertTrue(thrown.getValidationErrors().contains(guardMessage), + "ValidationException should carry the guard's message; was: " + thrown.getValidationErrors()); + } + + @Test + void testVerifyUpdateChecksDestinationGuardWhenDestinationChanges() { + stubRootConnectionDestination("current-destination-id"); + + assertDoesNotThrow(() -> connectionDAO.verifyUpdate(connectionDtoChangingDestination("new-destination-id"))); + verify(rootConnection).verifyCanUpdateDestination(); + } + + @Test + void testIsDestinationChangingReturnsFalseForSameDestinationId() { + final Connectable currentDestination = mock(Connectable.class); + when(currentDestination.getIdentifier()).thenReturn("dest-1"); + when(rootConnection.getDestination()).thenReturn(currentDestination); + + final ConnectableDTO proposed = new ConnectableDTO(); + proposed.setId("dest-1"); + proposed.setType(ConnectableType.PROCESSOR.name()); + + assertFalse(connectionDAO.isDestinationChanging(rootConnection, proposed)); + } + + @Test + void testIsDestinationChangingReturnsTrueForDifferentDestinationId() { + final Connectable currentDestination = mock(Connectable.class); + when(currentDestination.getIdentifier()).thenReturn("dest-1"); + when(rootConnection.getDestination()).thenReturn(currentDestination); + + final ConnectableDTO proposed = new ConnectableDTO(); + proposed.setId("dest-2"); + proposed.setType(ConnectableType.PROCESSOR.name()); + + assertTrue(connectionDAO.isDestinationChanging(rootConnection, proposed)); + } + + @Test + void testIsDestinationChangingRemoteInputPortSameGroupIsNotChanging() { + final RemoteGroupPort currentRemotePort = mock(RemoteGroupPort.class); + final RemoteProcessGroup currentRpg = mock(RemoteProcessGroup.class); + when(currentRemotePort.getIdentifier()).thenReturn("port-1"); + when(currentRemotePort.getConnectableType()).thenReturn(ConnectableType.REMOTE_INPUT_PORT); + when(currentRemotePort.getRemoteProcessGroup()).thenReturn(currentRpg); + when(currentRpg.getIdentifier()).thenReturn("rpg-A"); + when(rootConnection.getDestination()).thenReturn(currentRemotePort); + + final ConnectableDTO proposed = new ConnectableDTO(); + proposed.setId("port-1"); + proposed.setType(ConnectableType.REMOTE_INPUT_PORT.name()); + proposed.setGroupId("rpg-A"); + + assertFalse(connectionDAO.isDestinationChanging(rootConnection, proposed)); + } + + @Test + void testIsDestinationChangingRemoteInputPortDifferentGroupIsChanging() { + final RemoteGroupPort currentRemotePort = mock(RemoteGroupPort.class); + final RemoteProcessGroup currentRpg = mock(RemoteProcessGroup.class); + when(currentRemotePort.getIdentifier()).thenReturn("port-1"); + when(currentRemotePort.getConnectableType()).thenReturn(ConnectableType.REMOTE_INPUT_PORT); + when(currentRemotePort.getRemoteProcessGroup()).thenReturn(currentRpg); + when(currentRpg.getIdentifier()).thenReturn("rpg-A"); + when(rootConnection.getDestination()).thenReturn(currentRemotePort); + + final ConnectableDTO proposed = new ConnectableDTO(); + proposed.setId("port-1"); + proposed.setType(ConnectableType.REMOTE_INPUT_PORT.name()); + proposed.setGroupId("rpg-B"); + + assertTrue(connectionDAO.isDestinationChanging(rootConnection, proposed)); + } }