This is an automated email from the ASF dual-hosted git repository.
pvillard31 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 93caeedd42a NIFI-16476 Fix last queue dates for queues with many empty
remote partitions and priority queues (#11821)
93caeedd42a is described below
commit 93caeedd42a751e333089e28cd9040aa9316671e
Author: boguszj <[email protected]>
AuthorDate: Wed Oct 21 13:12:04 2026 +0200
NIFI-16476 Fix last queue dates for queues with many empty remote
partitions and priority queues (#11821)
---
.../nifi/controller/queue/SwappablePriorityQueue.java | 13 +++++++------
.../clustered/SocketLoadBalancedFlowFileQueue.java | 8 ++++++--
.../clustered/TestSocketLoadBalancedFlowFileQueue.java | 17 +++++++++++++++++
.../queue/clustered/TestSwappablePriorityQueue.java | 13 +++++++++++++
4 files changed, 43 insertions(+), 8 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/SwappablePriorityQueue.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/SwappablePriorityQueue.java
index dee5c139bf5..8778683713e 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/SwappablePriorityQueue.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/SwappablePriorityQueue.java
@@ -1269,7 +1269,7 @@ public class SwappablePriorityQueue {
try {
// We want the oldest timestamp, which will be the min
long min = getMinLastQueueDate(activeQueue, 0L);
- min = Long.min(min, getMinLastQueueDate(swapQueue, min));
+ min = getMinLastQueueDate(swapQueue, min);
for (Long minSwapQueueDate: minQueueDateInSwapLocation.values()) {
min = min == 0 ? minSwapQueueDate : Long.min(min,
minSwapQueueDate);
@@ -1281,14 +1281,15 @@ public class SwappablePriorityQueue {
}
}
- private long getMinLastQueueDate(Iterable<FlowFileRecord> iterable, long
defaultMin) {
- long min = 0;
+ private long getMinLastQueueDate(final Iterable<FlowFileRecord> iterable,
final long initialMin) {
+ long min = initialMin;
- for (FlowFileRecord flowFileRecord : iterable) {
- min = min == 0 ? flowFileRecord.getLastQueueDate() :
Long.min(flowFileRecord.getLastQueueDate(), min);
+ for (final FlowFileRecord flowFileRecord : iterable) {
+ final long lastQueueDate = flowFileRecord.getLastQueueDate();
+ min = min == 0 ? lastQueueDate : Long.min(lastQueueDate, min);
}
- return min == 0 ? defaultMin : min;
+ return min;
}
public long getTotalQueuedDuration(long fromTimestamp) {
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
index 4985cd58fe9..dde99caaabd 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
@@ -602,9 +602,13 @@ public class SocketLoadBalancedFlowFileQueue extends
AbstractFlowFileQueue imple
@Override
public long getMinLastQueueDate() {
long min = 0;
- for (QueuePartition queuePartition : queuePartitions) {
- min = min == 0 ? queuePartition.getMinLastQueueDate() :
Long.min(min, queuePartition.getMinLastQueueDate());
+ for (final QueuePartition queuePartition : queuePartitions) {
+ final long partitionMin = queuePartition.getMinLastQueueDate();
+ if (partitionMin > 0) {
+ min = min == 0 ? partitionMin : Long.min(min, partitionMin);
+ }
}
+
return min;
}
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 6c683ec9d5b..29c5353aebc 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
@@ -275,6 +275,23 @@ public class TestSocketLoadBalancedFlowFileQueue {
}
}
+ @Test
+ public void testMinLastQueueDateWithEmptyRemotePartitions() {
+ final MockFlowFileRecord localFlowFile = new MockFlowFileRecord();
+ queue.put(localFlowFile);
+
+ final QueuePartition localPartition = queue.getLocalPartition();
+ assertEquals(1, localPartition.size().getObjectCount());
+ for (int i = 0; i < nodeIds.size(); i++) {
+ final QueuePartition partition = queue.getPartition(i);
+ if (partition != localPartition) {
+ assertEquals(0, partition.size().getObjectCount());
+ }
+ }
+
+ assertEquals(localFlowFile.getLastQueueDate().longValue(),
queue.getMinLastQueueDate());
+ }
+
@Test
public void testPutAllBinsFlowFilesSeparately() {
// Partition data based on size. FlowFiles with 0 bytes will go to
partition 0 (local partition),
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/queue/clustered/TestSwappablePriorityQueue.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/queue/clustered/TestSwappablePriorityQueue.java
index 822e5406d90..1b0514ef64e 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/queue/clustered/TestSwappablePriorityQueue.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/queue/clustered/TestSwappablePriorityQueue.java
@@ -976,4 +976,17 @@ public class TestSwappablePriorityQueue {
assertEquals(totalNow - totalQueueDate,
queue.getTotalQueuedDuration(now));
assertEquals(minQueueDate, queue.getMinLastQueueDate());
}
+
+ @Test
+ public void testMinLastQueueDateWithEmptyActiveQueue() {
+ queue = new SwappablePriorityQueue(swapManager, 2, eventReporter,
flowFileQueue, dropAction, "local");
+ queue.put(new MockFlowFileRecord());
+ queue.put(new MockFlowFileRecord());
+
+ final MockFlowFileRecord swapQueueFlowFile = new MockFlowFileRecord();
+ queue.put(swapQueueFlowFile);
+ queue.poll(2, Collections.emptySet(), -1);
+
+ assertEquals(swapQueueFlowFile.getLastQueueDate().longValue(),
queue.getMinLastQueueDate());
+ }
}
