From 36edc2f80cf326f137b2a571419a9185bce1277d Mon Sep 17 00:00:00 2001 From: Mark Payne Date: Mon, 14 Sep 2026 19:33:59 -0400 Subject: [PATCH 1/3] NIFI-16344 Preserve Swap File Ordering during Selective Drop --- .../queue/SwappablePriorityQueue.java | 17 +++++++++-- .../clustered/TestSwappablePriorityQueue.java | 28 +++++++++++++++++++ 2 files changed, 43 insertions(+), 2 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 d2f13071c792..1542bc819a08 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 @@ -1014,11 +1014,9 @@ public SelectiveDropResult dropFlowFiles(final Predicate predicate) th final String newSwapLocation = result.newSwapLocation(); swapLocationUpdates.put(oldSwapLocation, newSwapLocation); - swapLocations.remove(oldSwapLocation); if (newSwapLocation != null) { // Some FlowFiles remain in new swap file - swapLocations.add(newSwapLocation); incrementSwapQueueSize(-result.droppedFlowFiles().size(), -result.droppedBytes(), 0); // Update metrics for the new swap location @@ -1040,6 +1038,21 @@ public SelectiveDropResult dropFlowFiles(final Predicate predicate) th droppedFlowFiles.addAll(result.droppedFlowFiles()); } + // Replace rewritten swap files in place so that retained FlowFiles remain ahead of later swap files. + final List updatedSwapLocations = new ArrayList<>(swapLocations.size()); + for (final String swapLocation : swapLocations) { + if (swapLocationUpdates.containsKey(swapLocation)) { + final String newSwapLocation = swapLocationUpdates.get(swapLocation); + if (newSwapLocation != null) { + updatedSwapLocations.add(newSwapLocation); + } + } else { + updatedSwapLocations.add(swapLocation); + } + } + swapLocations.clear(); + swapLocations.addAll(updatedSwapLocations); + // Filter the active queue final Queue newActiveQueue = new PriorityQueue<>(Math.max(20, activeQueue.size()), new QueuePrioritizer(getPriorities())); int droppedFromActiveCount = 0; 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 f8035fd24510..822e5406d907 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 @@ -734,6 +734,34 @@ public void testSelectiveDropFromSwapFile() throws IOException { assertEquals(2, result.getSwapLocationUpdates().size()); } + @Test + public void testSelectiveDropPreservesSwapFileOrdering() throws IOException { + queue = new SwappablePriorityQueue(swapManager, 10, eventReporter, flowFileQueue, dropAction, "local"); + + for (int i = 0; i < 30; i++) { + queue.put(new MockFlowFileRecord(Map.of("index", Integer.toString(i)), 1L)); + } + + assertEquals(2, swapManager.swappedOut.size()); + + final SelectiveDropResult result = queue.dropFlowFiles(flowFile -> "10".equals(flowFile.getAttribute("index"))); + + assertEquals(1, result.getDroppedCount()); + assertEquals(1, result.getSwapLocationUpdates().size()); + + final Set expired = new HashSet<>(); + for (int expectedIndex = 0; expectedIndex < 30; expectedIndex++) { + if (expectedIndex == 10) { + continue; + } + + final FlowFileRecord flowFile = queue.poll(expired, 0L); + assertNotNull(flowFile); + assertEquals(Integer.toString(expectedIndex), flowFile.getAttribute("index")); + } + assertNull(queue.poll(expired, 0L)); + } + @Test @Timeout(120) public void testSelectiveDropRemovesEntireSwapFileWhenAllMatch() throws IOException { From 966837584b9e0d612368452adc29cd0459d87d01 Mon Sep 17 00:00:00 2001 From: Mark Payne Date: Tue, 15 Sep 2026 11:57:06 -0400 Subject: [PATCH 2/3] Update nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/SwappablePriorityQueue.java Co-authored-by: Bob Paulin --- .../apache/nifi/controller/queue/SwappablePriorityQueue.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 1542bc819a08..0805e3b6dcb3 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 @@ -1039,7 +1039,7 @@ public SelectiveDropResult dropFlowFiles(final Predicate predicate) th } // Replace rewritten swap files in place so that retained FlowFiles remain ahead of later swap files. - final List updatedSwapLocations = new ArrayList<>(swapLocations.size()); + final List rewrittenSwapLocations = new ArrayList<>(swapLocations.size()); for (final String swapLocation : swapLocations) { if (swapLocationUpdates.containsKey(swapLocation)) { final String newSwapLocation = swapLocationUpdates.get(swapLocation); From 28249a35ad108ac56c30b977ce2c077f6544255f Mon Sep 17 00:00:00 2001 From: Mark Payne Date: Tue, 15 Sep 2026 12:02:16 -0400 Subject: [PATCH 3/3] NIFI-16344 Complete rewrittenSwapLocations rename in selective drop Update remaining references in dropFlowFiles after renaming the local variable per review feedback. Co-authored-by: Cursor --- .../nifi/controller/queue/SwappablePriorityQueue.java | 6 +++--- 1 file changed, 3 insertions(+), 3 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 0805e3b6dcb3..57f18cc76406 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 @@ -1044,14 +1044,14 @@ public SelectiveDropResult dropFlowFiles(final Predicate predicate) th if (swapLocationUpdates.containsKey(swapLocation)) { final String newSwapLocation = swapLocationUpdates.get(swapLocation); if (newSwapLocation != null) { - updatedSwapLocations.add(newSwapLocation); + rewrittenSwapLocations.add(newSwapLocation); } } else { - updatedSwapLocations.add(swapLocation); + rewrittenSwapLocations.add(swapLocation); } } swapLocations.clear(); - swapLocations.addAll(updatedSwapLocations); + swapLocations.addAll(rewrittenSwapLocations); // Filter the active queue final Queue newActiveQueue = new PriorityQueue<>(Math.max(20, activeQueue.size()), new QueuePrioritizer(getPriorities()));