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 9d6992fa1df NIFI-16180 Impoved unstable test for FlowFile Queue
rebalancing (#11568)
9d6992fa1df is described below
commit 9d6992fa1df6170ab877c145f36fe90c3b21f2b1
Author: David Handermann <[email protected]>
AuthorDate: Wed Aug 19 17:49:07 2026 -0500
NIFI-16180 Impoved unstable test for FlowFile Queue rebalancing (#11568)
---
.../TestSocketLoadBalancedFlowFileQueue.java | 33 ++++++++++++++++++++--
1 file changed, 31 insertions(+), 2 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/queue/clustered/TestSocketLoadBalancedFlowFileQueue.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/queue/clustered/TestSocketLoadBalancedFlowFileQueue.java
index 33cc02bf09f..cfa0898faa6 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/queue/clustered/TestSocketLoadBalancedFlowFileQueue.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/queue/clustered/TestSocketLoadBalancedFlowFileQueue.java
@@ -78,6 +78,13 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
public class TestSocketLoadBalancedFlowFileQueue {
+ /**
+ * Prefix of the name that {@code StandardRebalancingPartition} assigns to
the thread that redistributes
+ * FlowFiles held by the rebalancing partition. The remainder of the name
is the queue identifier.
+ */
+ private static final String REBALANCE_THREAD_NAME_PREFIX = "Rebalance
queued data for Connection ";
+ private static final long REBALANCE_TERMINATION_TIMEOUT = 10_000L;
+ private static final long REBALANCE_TERMINATION_POLL_INTERVAL = 10L;
private FlowFileRepository flowFileRepo;
private ContentRepository contentRepo;
@@ -764,16 +771,26 @@ public class TestSocketLoadBalancedFlowFileQueue {
}
@Test
- public void testGetQueueSnapshotIncludesRebalancingPartition() {
+ @Timeout(30)
+ public void testGetQueueSnapshotIncludesRebalancingPartition() throws
InterruptedException {
// Route FlowFiles to the local partition, then move every partition's
contents into the rebalancing
// partition and prevent it from redistributing them. FlowFiles
sitting in the rebalancing partition
// (being redistributed across the cluster) must still be counted in
the snapshot's total QueueSize.
final int localPartitionIndex = determineLocalPartitionIndex();
queue.setFlowFilePartitioner(new
StaticFlowFilePartitioner(localPartitionIndex));
- // Toggle load balancing so the rebalancing partition is stopped and
will not drain FlowFiles moved into it.
+ // The rebalancing partition runs a background thread from the time
the queue is created, so load balancing
+ // must be started and stopped to mark the rebalancing partition as
stopped. The background thread observes
+ // the stopped state only after its in-flight poll returns, and it
redistributes whatever that poll returns.
+ // FlowFiles are therefore only guaranteed to remain in the
rebalancing partition once the thread terminates.
queue.startLoadBalancing();
queue.stopLoadBalancing();
+ final long terminationExpiration = System.currentTimeMillis() +
REBALANCE_TERMINATION_TIMEOUT;
+ while (isRebalanceThreadRunning() && System.currentTimeMillis() <
terminationExpiration) {
+ Thread.sleep(REBALANCE_TERMINATION_POLL_INTERVAL);
+ }
+
+ assertFalse(isRebalanceThreadRunning(), "Rebalancing Partition thread
not terminated");
final long bytesPerFlowFile = 5L;
final int flowFileCount = 4;
@@ -795,6 +812,18 @@ public class TestSocketLoadBalancedFlowFileQueue {
assertTrue(snapshot.activeFlowFiles().isEmpty());
}
+ private boolean isRebalanceThreadRunning() {
+ final String rebalanceThreadName = REBALANCE_THREAD_NAME_PREFIX +
queue.getIdentifier();
+
+ for (final Thread thread : Thread.getAllStackTraces().keySet()) {
+ if (thread.isAlive() &&
rebalanceThreadName.equals(thread.getName())) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
private void assertPartitionSizes(final int[] expectedSizes) {
final int[] partitionSizes = new int[queue.getPartitionCount()];
while (!Arrays.equals(expectedSizes, partitionSizes)) {
