NIFI-15903 Improved flaky tests ClusteredConnectorIT and ContentClaimTruncationIT (apache#11203)

- Added test method for robust cluster node removal based on coordinator reported status

Signed-off-by: David Handermann <exceptionfactory@apache.org>

Author: pvillard31

nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java

@@ -1171,6 +1171,23 @@ public void waitForProcessorState(final String processorId, final String expecte
             final Integer terminatedThreadCount = snapshotDto.getTerminatedThreadCount();
 
             if ("RUNNING".equals(expectedState) || (activeThreadCount == 0 && terminatedThreadCount == 0)) {
+                // The logical state masks the framework's physical STOPPING state as STOPPED. The framework's
+                // verifyCanStart check evaluates the physical state, so when the caller is waiting for STOPPED
+                // we additionally require the physical state to have settled to STOPPED or DISABLED. Without
+                // this, a tight loop of runProcessorOnce + waitForStoppedProcessor can race against an
+                // in-flight stop transition and the next start request fails with "cannot be started because
+                // it is not stopped. Current state is STOPPING".
+                if ("STOPPED".equalsIgnoreCase(expectedState)) {
+                    final String physicalState = entity.getComponent().getPhysicalState();
+                    final boolean physicalStateSettled = physicalState == null
+                            || "STOPPED".equalsIgnoreCase(physicalState)
+                            || "DISABLED".equalsIgnoreCase(physicalState);
+                    if (!physicalStateSettled) {
+                        Thread.sleep(10L);
+                        continue;
+                    }
+                }
+
                 logger.info("Processor {} is now in desired state of {} with {} active threads and {} terminated threads",
                     processorId, expectedState, activeThreadCount, terminatedThreadCount);
                 return;

nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiSystemIT.java

@@ -703,6 +703,30 @@ protected void reconnectNode(final int nodeIndex) throws NiFiClientException, IO
         getNifiClient().getControllerClient().connectNode(nodeEntity.getNode().getNodeId(), nodeEntity);
     }
 
+    /**
+     * Removes a node from the cluster and waits until the cluster coordinator no longer reports the node.
+     * Mutating cluster requests (such as deleting a component) reject while a non-CONNECTED node is still
+     * known to the coordinator, so callers that need to issue follow-up mutating requests must wait for
+     * the removal to be reflected in the cluster state rather than relying on the synchronous response of
+     * {@code deleteNode} alone.
+     *
+     * @param nodeIndex the 1-based index of the node
+     */
+    protected void removeNodeFromCluster(final int nodeIndex) throws NiFiClientException, IOException, InterruptedException {
+        final NodeEntity nodeEntity = getNodeEntity(nodeIndex);
+        final int expectedPort = getClientApiPort() + nodeIndex - 1;
+        getNifiClient().getControllerClient().deleteNode(nodeEntity.getNode().getNodeId());
+
+        waitFor(() -> {
+            try {
+                return getNifiClient().getControllerClient().getNodes().getCluster().getNodes().stream()
+                        .noneMatch(dto -> dto.getApiPort() == expectedPort);
+            } catch (final Exception e) {
+                return false;
+            }
+        });
+    }
+
     protected NodeEntity getNodeEntity(final int nodeIndex) throws NiFiClientException, IOException {
         final ClusterEntity clusterEntity = getNifiClient().getControllerClient().getNodes();
         final int expectedPort = getClientApiPort() + nodeIndex - 1;

nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/JoinClusterWithMissingConnectionWithDataIT.java

@@ -99,8 +99,7 @@ public void testFailsToJoinWithMissingConnectionThatHasData() throws NiFiClientE
         node2.stop();
 
         // Remove node from the cluster
-        getNifiClient().getControllerClient().deleteNode(node2Dto.getNodeId());
-        waitFor(() -> isNodeRemoved(5672));
+        removeNodeFromCluster(2);
 
         // Drop the data in the queue and delete the queue.
         getClientUtil().emptyQueue(CONNECTION_UUID);
@@ -113,15 +112,6 @@ public void testFailsToJoinWithMissingConnectionThatHasData() throws NiFiClientE
         waitFor(() -> isNodeDisconnected(5672));
     }
 
-    private boolean isNodeRemoved(final int apiPort) {
-        try {
-            return getNifiClient().getControllerClient().getNodes().getCluster().getNodes().stream()
-                .noneMatch(dto -> dto.getApiPort() == apiPort);
-        } catch (Exception e) {
-            return false;
-        }
-    }
-
     private boolean isNodeDisconnected(final int apiPort) {
         try {
             final NodeDTO nodeDto = getNifiClient().getControllerClient().getNodes().getCluster().getNodes().stream()

nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ClusteredConnectorIT.java

@@ -24,7 +24,6 @@
 import org.apache.nifi.web.api.dto.ConnectorConfigurationDTO;
 import org.apache.nifi.web.api.dto.ConnectorValueReferenceDTO;
 import org.apache.nifi.web.api.entity.ConnectorEntity;
-import org.apache.nifi.web.api.entity.NodeEntity;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -130,8 +129,7 @@ public void testDeleteConnectorOnConnect() throws NiFiClientException, IOExcepti
         final ConnectorClient connectorClient = getNifiClient().getConnectorClient();
         assertThrows(NiFiClientException.class, () -> connectorClient.deleteConnector(connector));
 
-        final NodeEntity node2Entity = getNodeEntity(2);
-        getNifiClient().getControllerClient().deleteNode(node2Entity.getNode().getNodeId());
+        removeNodeFromCluster(2);
 
         // Should now be able to delete connector
         connectorClient.deleteConnector(connector);
@@ -172,8 +170,7 @@ public void testDeleteConnectorOnConnectWithDataQueued() throws NiFiClientExcept
         assertThrows(NiFiClientException.class, () -> connectorClient.deleteConnector(connector));
 
         // Remove node 2 from cluster.
-        final NodeEntity node2Entity = getNodeEntity(2);
-        getNifiClient().getControllerClient().deleteNode(node2Entity.getNode().getNodeId());
+        removeNodeFromCluster(2);
 
         // We cannot delete the connector directly because it has data queued. Stop Node 1, delete the flow.json.gz file, and restart Node 1.
         getNiFiInstance().getNodeInstance(1).stop();