NIFI-16082 - Connection update is incorrectly rejected when the current destination is running

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

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

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

nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardConnectionDAO.java

@@ -519,19 +519,49 @@ 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
             connection.verifyCanUpdate();
         }
     }
 
+    /**
+     * 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,26 +601,15 @@ 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
                 if (proposedDestination.getGroupId() == null) {
                     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());