NIFI-11159 Fixing connections with source having reassigned id

pvillard31 · Feb 10, 2023 · b0ec28a · b0ec28a
1 parent a66727c
commit b0ec28a
Showing 1 changed file with 23 additions and 0 deletions.
diff --git a/...ain/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java b/...ain/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
@@ -614,16 +614,39 @@ private void synchronizeControllerServices(final ProcessGroup group, final Versi
 
     private void removeMissingConnections(final ProcessGroup group, final VersionedProcessGroup proposed, final Map<String, Connection> connectionsByVersionedId) {
         final Set<String> connectionsRemoved = new HashSet<>(connectionsByVersionedId.keySet());
+        final Set<String> connectionsRemovedDueToChangingSourceId = new HashSet<>();
 
         for (final VersionedConnection proposedConnection : proposed.getConnections()) {
             connectionsRemoved.remove(proposedConnection.getIdentifier());
         }
 
+        // Check for any case where there's an existing connection whose ID matches the proposed connection, but whose source doesn't match
+        // the proposed source ID. The source of a Connection should never change from one component to another. However, there are cases
+        // in which the Versioned Component ID might change, in order to avoid conflicts with sibling Process Groups. In such a case, we must remove
+        // the connection and create a new one, since we cannot simply change the source in the same way that we can change the destination.
+        for (final VersionedConnection proposedConnection : proposed.getConnections()) {
+            final Connection existingConnection = connectionsByVersionedId.get(proposedConnection.getIdentifier());
+
+            if (existingConnection != null) {
+                final String proposedSourceId = proposedConnection.getSource().getId();
+                final String existingSourceId = existingConnection.getSource().getVersionedComponentId().orElse(null);
+
+                if (!Objects.equals(proposedSourceId, existingSourceId)) {
+                    connectionsRemovedDueToChangingSourceId.add(proposedConnection.getIdentifier());
+                    connectionsRemoved.add(proposedConnection.getIdentifier());
+                }
+            }
+        }
+
         for (final String removedVersionedId : connectionsRemoved) {
             final Connection connection = connectionsByVersionedId.get(removedVersionedId);
             LOG.info("Removing {} from {}", connection, group);
             group.removeConnection(connection);
         }
+
+        for (final String removedVersionedId : connectionsRemovedDueToChangingSourceId) {
+            connectionsByVersionedId.remove(removedVersionedId);
+        }
     }
 
     private void synchronizeConnections(final ProcessGroup group, final VersionedProcessGroup proposed, final Map<String, Connection> connectionsByVersionedId) {