NIFI-16232 - Allow versioned flow updates to drain removed connections

nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-integration-tests/src/test/java/org/apache/nifi/mock/connectors/tests/CreateConnectorIT.java

@@ -42,7 +42,7 @@
 public class CreateConnectorIT {
 
     @Test
-    public void testCreateStartAndStopGenerateAndUpdateConnector() throws IOException {
+    public void testCreateStartAndStopGenerateAndUpdateConnector() throws IOException, TimeoutException {
         try (final ConnectorTestRunner testRunner = new StandardConnectorTestRunner.Builder()
             .connectorClassName("org.apache.nifi.mock.connectors.GenerateAndLog")
             .narLibraryDirectory(new File("target/libDir"))
@@ -65,7 +65,7 @@ public void testCreateStartAndStopGenerateAndUpdateConnector() throws IOExceptio
             assertEquals("org.apache.nifi.lookup.SimpleKeyValueLookupService", controllerServices.iterator().next().getType());
 
             testRunner.startConnector();
-            testRunner.stopConnector();
+            testRunner.stopConnector(Duration.ofSeconds(120));
         }
     }
 
@@ -77,7 +77,6 @@ public void testStopConnectorWithTimeoutStopsRunningConnector() throws IOExcepti
                 .build()) {
 
             testRunner.startConnector();
-
             testRunner.stopConnector(Duration.ofSeconds(120));
         }
     }

nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/FlowUpdateImpact.java

@@ -0,0 +1,85 @@
+/*
+ * 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.web;
+
+import org.apache.nifi.web.api.entity.AffectedComponentEntity;
+
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.Objects;
+import java.util.Set;
+
+public final class FlowUpdateImpact {
+    private final Set<AffectedComponentEntity> affectedComponents;
+    private final Set<RemovedConnectionDescriptor> removedConnections;
+    private final Set<String> removedProcessGroupIds;
+    private final Set<String> removedEndpointIds;
+
+    public FlowUpdateImpact(final Set<AffectedComponentEntity> affectedComponents,
+                            final Set<RemovedConnectionDescriptor> removedConnections,
+                            final Set<String> removedProcessGroupIds,
+                            final Set<String> removedEndpointIds) {
+        this.affectedComponents = unmodifiableCopy(affectedComponents);
+        this.removedConnections = unmodifiableCopy(removedConnections);
+        this.removedProcessGroupIds = unmodifiableCopy(removedProcessGroupIds);
+        this.removedEndpointIds = unmodifiableCopy(removedEndpointIds);
+    }
+
+    public Set<AffectedComponentEntity> getAffectedComponents() {
+        return affectedComponents;
+    }
+
+    public Set<RemovedConnectionDescriptor> getRemovedConnections() {
+        return removedConnections;
+    }
+
+    public Set<String> getRemovedProcessGroupIds() {
+        return removedProcessGroupIds;
+    }
+
+    public Set<String> getRemovedEndpointIds() {
+        return removedEndpointIds;
+    }
+
+    private static <T> Set<T> unmodifiableCopy(final Set<T> values) {
+        if (values == null || values.isEmpty()) {
+            return Collections.emptySet();
+        }
+
+        return Collections.unmodifiableSet(new LinkedHashSet<>(values));
+    }
+
+    @Override
+    public boolean equals(final Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (!(obj instanceof final FlowUpdateImpact other)) {
+            return false;
+        }
+
+        return Objects.equals(affectedComponents, other.affectedComponents)
+                && Objects.equals(removedConnections, other.removedConnections)
+                && Objects.equals(removedProcessGroupIds, other.removedProcessGroupIds)
+                && Objects.equals(removedEndpointIds, other.removedEndpointIds);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(affectedComponents, removedConnections, removedProcessGroupIds, removedEndpointIds);
+    }
+}

nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacade.java

@@ -2079,6 +2079,24 @@ VersionControlInformationEntity setVersionControlInformation(Revision processGro
      */
     String getFlowRegistryName(String flowRegistryId);
 
+    /**
+     * Determines which components currently exist in the Process Group with the given identifier and calculates which of those components
+     * would be impacted by updating the Process Group to the provided snapshot
+     *
+     * @param processGroupId the ID of the Process Group to update
+     * @param updatedSnapshot the snapshot to update the Process Group to
+     * @return the impact of updating the Process Group
+     */
+    FlowUpdateImpact getFlowUpdateImpact(String processGroupId, RegisteredFlowSnapshot updatedSnapshot);
+
+    /**
+     * Returns a live view of connections, connectables, and Process Groups used to classify removed connections.
+     * Callers must read queue and component state again when needed; the view does not snapshot runtime state.
+     *
+     * @return the current flow context for removed-connection drain classification
+     */
+    RemovedConnectionDrainClassifier.Context getRemovedConnectionDrainContext();
+
     /**
      * Determines which components currently exist in the Process Group with the given identifier and calculates which of those components
      * would be impacted by updating the Process Group to the provided snapshot

nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/RemovalReason.java

@@ -0,0 +1,22 @@
+/*
+ * 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.web;
+
+public enum RemovalReason {
+    COMPONENT_REMOVED,
+    SOURCE_CHANGED
+}

nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/RemovedConnectionDescriptor.java

@@ -0,0 +1,136 @@
+/*
+ * 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.web;
+
+import org.apache.nifi.connectable.ConnectableType;
+
+import java.util.Objects;
+
+public final class RemovedConnectionDescriptor {
+    private final String connectionInstanceId;
+    private final String connectionVersionedId;
+    private final String containingProcessGroupId;
+    private final String sourceInstanceId;
+    private final String sourceVersionedId;
+    private final String sourceProcessGroupId;
+    private final ConnectableType sourceType;
+    private final String destinationInstanceId;
+    private final String destinationVersionedId;
+    private final String destinationProcessGroupId;
+    private final ConnectableType destinationType;
+    private final RemovalReason removalReason;
+
+    public RemovedConnectionDescriptor(final String connectionInstanceId, final String connectionVersionedId,
+                                       final String containingProcessGroupId,
+                                       final String sourceInstanceId, final String sourceVersionedId, final String sourceProcessGroupId,
+                                       final ConnectableType sourceType,
+                                       final String destinationInstanceId, final String destinationVersionedId,
+                                       final String destinationProcessGroupId, final ConnectableType destinationType,
+                                       final RemovalReason removalReason) {
+        this.connectionInstanceId = connectionInstanceId;
+        this.connectionVersionedId = connectionVersionedId;
+        this.containingProcessGroupId = containingProcessGroupId;
+        this.sourceInstanceId = sourceInstanceId;
+        this.sourceVersionedId = sourceVersionedId;
+        this.sourceProcessGroupId = sourceProcessGroupId;
+        this.sourceType = sourceType;
+        this.destinationInstanceId = destinationInstanceId;
+        this.destinationVersionedId = destinationVersionedId;
+        this.destinationProcessGroupId = destinationProcessGroupId;
+        this.destinationType = destinationType;
+        this.removalReason = removalReason;
+    }
+
+    public String getConnectionInstanceId() {
+        return connectionInstanceId;
+    }
+
+    public String getConnectionVersionedId() {
+        return connectionVersionedId;
+    }
+
+    public String getContainingProcessGroupId() {
+        return containingProcessGroupId;
+    }
+
+    public String getSourceInstanceId() {
+        return sourceInstanceId;
+    }
+
+    public String getSourceVersionedId() {
+        return sourceVersionedId;
+    }
+
+    public String getSourceProcessGroupId() {
+        return sourceProcessGroupId;
+    }
+
+    public ConnectableType getSourceType() {
+        return sourceType;
+    }
+
+    public String getDestinationInstanceId() {
+        return destinationInstanceId;
+    }
+
+    public String getDestinationVersionedId() {
+        return destinationVersionedId;
+    }
+
+    public String getDestinationProcessGroupId() {
+        return destinationProcessGroupId;
+    }
+
+    public ConnectableType getDestinationType() {
+        return destinationType;
+    }
+
+    public RemovalReason getRemovalReason() {
+        return removalReason;
+    }
+
+    @Override
+    public boolean equals(final Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (!(obj instanceof final RemovedConnectionDescriptor other)) {
+            return false;
+        }
+
+        return Objects.equals(connectionInstanceId, other.connectionInstanceId)
+                && Objects.equals(connectionVersionedId, other.connectionVersionedId)
+                && Objects.equals(containingProcessGroupId, other.containingProcessGroupId)
+                && Objects.equals(sourceInstanceId, other.sourceInstanceId)
+                && Objects.equals(sourceVersionedId, other.sourceVersionedId)
+                && Objects.equals(sourceProcessGroupId, other.sourceProcessGroupId)
+                && sourceType == other.sourceType
+                && Objects.equals(destinationInstanceId, other.destinationInstanceId)
+                && Objects.equals(destinationVersionedId, other.destinationVersionedId)
+                && Objects.equals(destinationProcessGroupId, other.destinationProcessGroupId)
+                && destinationType == other.destinationType
+                && removalReason == other.removalReason;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(connectionInstanceId, connectionVersionedId, containingProcessGroupId,
+                sourceInstanceId, sourceVersionedId, sourceProcessGroupId, sourceType,
+                destinationInstanceId, destinationVersionedId, destinationProcessGroupId, destinationType,
+                removalReason);
+    }
+}