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 2f74b170765 NIFI-15903 Improved flaky tests ClusteredConnectorIT and
ContentClaimTruncationIT (#11203)
2f74b170765 is described below
commit 2f74b1707652a5dfb615cb6997a3d93deb81db2c
Author: Pierre Villard <[email protected]>
AuthorDate: Tue May 5 15:47:44 2026 +0200
NIFI-15903 Improved flaky tests ClusteredConnectorIT and
ContentClaimTruncationIT (#11203)
- Added test method for robust cluster node removal based on coordinator
reported status
Signed-off-by: David Handermann <[email protected]>
---
.../apache/nifi/tests/system/NiFiClientUtil.java | 17 +++++++++++++++
.../org/apache/nifi/tests/system/NiFiSystemIT.java | 24 ++++++++++++++++++++++
...JoinClusterWithMissingConnectionWithDataIT.java | 12 +----------
.../system/connectors/ClusteredConnectorIT.java | 7 ++-----
4 files changed, 44 insertions(+), 16 deletions(-)
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java
index 17de5d0ce14..2de5c0f915e 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java
@@ -1171,6 +1171,23 @@ public class NiFiClientUtil {
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;
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiSystemIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiSystemIT.java
index fab72e3cd50..fae7e7bd76e 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiSystemIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiSystemIT.java
@@ -703,6 +703,30 @@ public abstract class NiFiSystemIT implements
NiFiInstanceProvider {
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;
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/JoinClusterWithMissingConnectionWithDataIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/JoinClusterWithMissingConnectionWithDataIT.java
index 55127a49e36..d6bb11603d1 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/JoinClusterWithMissingConnectionWithDataIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/JoinClusterWithMissingConnectionWithDataIT.java
@@ -99,8 +99,7 @@ public class JoinClusterWithMissingConnectionWithDataIT
extends NiFiSystemIT {
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 class JoinClusterWithMissingConnectionWithDataIT
extends NiFiSystemIT {
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()
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ClusteredConnectorIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ClusteredConnectorIT.java
index cc22e9fa271..d377b866362 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ClusteredConnectorIT.java
+++
b/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.toolkit.client.NiFiClientException;
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 class ClusteredConnectorIT extends ConnectorCrudIT {
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 class ClusteredConnectorIT extends ConnectorCrudIT {
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();
