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 43b5b883d7e NIFI-16082 Fixed Connection update when the current
destination is running (#11400)
43b5b883d7e is described below
commit 43b5b883d7e65ef73a29572d1ebfde59896ad565
Author: Pierre Villard <[email protected]>
AuthorDate: Fri Aug 14 15:52:35 2026 +0200
NIFI-16082 Fixed Connection update when the current destination is running
(#11400)
Signed-off-by: David Handermann <[email protected]>
---
.../nifi/connectable/StandardConnection.java | 39 +++---
.../nifi/connectable/StandardConnectionTest.java | 129 +++++++++++++++++++
.../org/apache/nifi/connectable/Connection.java | 6 +
.../nifi/web/dao/impl/StandardConnectionDAO.java | 57 ++++++---
.../web/dao/impl/StandardConnectionDAOTest.java | 139 +++++++++++++++++++++
5 files changed, 337 insertions(+), 33 deletions(-)
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 fe712ab363d..96359e09ff0 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,31 @@ public final class StandardConnection implements
Connection {
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;
+ }
+ }
+
+ /**
+ * @throws IllegalStateException if the current destination is running and
is not exempt, or FlowFiles from this
+ * Connection are currently held by the destination
+ */
+ @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 +335,6 @@ public final class StandardConnection implements
Connection {
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 00000000000..e2f9d334973
--- /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 423f52d4cec..1b801576c33 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,12 @@ 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.
+ */
+ void verifyCanUpdateDestination();
+
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 a6d845d5f72..4a5691d7567 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 @@ public class StandardConnectionDAO extends ComponentDAO
implements ConnectionDAO
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 @@ public class StandardConnectionDAO extends ComponentDAO
implements ConnectionDAO
}
}
+ /**
+ * 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 class StandardConnectionDAO extends ComponentDAO
implements ConnectionDAO
// 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 class StandardConnectionDAO extends ComponentDAO
implements ConnectionDAO
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 class StandardConnectionDAO extends ComponentDAO
implements ConnectionDAO
}
} 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 f368ffc2e28..70e2e1a78d5 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.ConnectorRepository;
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,11 +44,15 @@ import org.mockito.quality.Strictness;
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.doThrow;
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)
@@ -81,6 +92,14 @@ class StandardConnectionDAOTest {
private static final String ROOT_CONNECTION_ID = "root-connection-id";
private static final String CONNECTOR_CONNECTION_ID =
"connector-connection-id";
private static final String NON_EXISTENT_ID = "non-existent-id";
+ private static final String PROCESS_GROUP_ID = "group-id";
+ private static final String CURRENT_DESTINATION_ID =
"current-destination-id";
+ private static final String NEW_DESTINATION_ID = "new-destination-id";
+ private static final String DESTINATION_ID = "dest-1";
+ private static final String ALTERNATIVE_DESTINATION_ID = "dest-2";
+ private static final String REMOTE_PORT_ID = "port-1";
+ private static final String REMOTE_PROCESS_GROUP_ID = "rpg-A";
+ private static final String ALTERNATIVE_REMOTE_PROCESS_GROUP_ID = "rpg-B";
@BeforeEach
void setUp() {
@@ -200,4 +219,124 @@ class StandardConnectionDAOTest {
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(PROCESS_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";
+ 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(DESTINATION_ID);
+ when(rootConnection.getDestination()).thenReturn(currentDestination);
+
+ final ConnectableDTO proposed = new ConnectableDTO();
+ proposed.setId(DESTINATION_ID);
+ proposed.setType(ConnectableType.PROCESSOR.name());
+
+ assertFalse(connectionDAO.isDestinationChanging(rootConnection,
proposed));
+ }
+
+ @Test
+ void testIsDestinationChangingReturnsTrueForDifferentDestinationId() {
+ final Connectable currentDestination = mock(Connectable.class);
+ when(currentDestination.getIdentifier()).thenReturn(DESTINATION_ID);
+ when(rootConnection.getDestination()).thenReturn(currentDestination);
+
+ final ConnectableDTO proposed = new ConnectableDTO();
+ proposed.setId(ALTERNATIVE_DESTINATION_ID);
+ 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(REMOTE_PORT_ID);
+
when(currentRemotePort.getConnectableType()).thenReturn(ConnectableType.REMOTE_INPUT_PORT);
+ when(currentRemotePort.getRemoteProcessGroup()).thenReturn(currentRpg);
+ when(currentRpg.getIdentifier()).thenReturn(REMOTE_PROCESS_GROUP_ID);
+ when(rootConnection.getDestination()).thenReturn(currentRemotePort);
+
+ final ConnectableDTO proposed = new ConnectableDTO();
+ proposed.setId(REMOTE_PORT_ID);
+ proposed.setType(ConnectableType.REMOTE_INPUT_PORT.name());
+ proposed.setGroupId(REMOTE_PROCESS_GROUP_ID);
+
+ assertFalse(connectionDAO.isDestinationChanging(rootConnection,
proposed));
+ }
+
+ @Test
+ void testIsDestinationChangingRemoteInputPortDifferentGroupIsChanging() {
+ final RemoteGroupPort currentRemotePort = mock(RemoteGroupPort.class);
+ final RemoteProcessGroup currentRpg = mock(RemoteProcessGroup.class);
+ when(currentRemotePort.getIdentifier()).thenReturn(REMOTE_PORT_ID);
+
when(currentRemotePort.getConnectableType()).thenReturn(ConnectableType.REMOTE_INPUT_PORT);
+ when(currentRemotePort.getRemoteProcessGroup()).thenReturn(currentRpg);
+ when(currentRpg.getIdentifier()).thenReturn(REMOTE_PROCESS_GROUP_ID);
+ when(rootConnection.getDestination()).thenReturn(currentRemotePort);
+
+ final ConnectableDTO proposed = new ConnectableDTO();
+ proposed.setId(REMOTE_PORT_ID);
+ proposed.setType(ConnectableType.REMOTE_INPUT_PORT.name());
+ proposed.setGroupId(ALTERNATIVE_REMOTE_PROCESS_GROUP_ID);
+
+ assertTrue(connectionDAO.isDestinationChanging(rootConnection,
proposed));
+ }
}
