markap14 commented on code in PR #11921:
URL: https://github.com/apache/nifi/pull/11921#discussion_r4568544215
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/SwappablePriorityQueue.java:
##########
@@ -1284,6 +1288,62 @@ public SwapSummary recoverSwappedFlowFiles() {
return new StandardSwapSummary(new QueueSize(swapFlowFileCount,
swapByteCount), maxId, resourceClaims, minSwappedQueueDate,
totalSwappedQueueDate);
}
+ /**
+ * On restart, the FlowFiles that were held in memory are restored from
the FlowFile Repository before the swap files are recovered, so they
+ * fill the active queue regardless of whether they were queued before or
after the swapped FlowFiles. Because swap files are not swapped in
+ * until the active queue is empty, this keeps in the active queue only
the FlowFiles that come before the highest priority FlowFile of the
+ * oldest swap file and moves all others to the swap queue, which is
processed after the swap files. The queue is placed in swap mode so that
+ * FlowFiles added afterward are also processed after the swap files.
+ *
+ * This method MUST be called with the write lock held.
+ *
+ * @param oldestSwapLocation the location of the oldest recovered swap file
+ */
+ private void restoreActiveFlowFilesToSwapQueue(final String
oldestSwapLocation) {
+ swapMode = true;
Review Comment:
[Claude Opus 5.5] Setting swap mode is intentionally independent of reading
the swap file. Whenever a queue has swap files, FlowFiles added afterward need
to go to the swap queue so that they are processed after the swap files. If
swap mode stayed off when the swap file could not be read, new FlowFiles would
go into the active queue ahead of the swapped FlowFiles. To make that clearer,
af957eec3cd moves the assignment into `recoverSwappedFlowFiles()`, next to the
check for recovered swap files, with a short comment explaining why.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/SwappablePriorityQueue.java:
##########
@@ -1284,6 +1288,62 @@ public SwapSummary recoverSwappedFlowFiles() {
return new StandardSwapSummary(new QueueSize(swapFlowFileCount,
swapByteCount), maxId, resourceClaims, minSwappedQueueDate,
totalSwappedQueueDate);
}
+ /**
+ * On restart, the FlowFiles that were held in memory are restored from
the FlowFile Repository before the swap files are recovered, so they
+ * fill the active queue regardless of whether they were queued before or
after the swapped FlowFiles. Because swap files are not swapped in
+ * until the active queue is empty, this keeps in the active queue only
the FlowFiles that come before the highest priority FlowFile of the
+ * oldest swap file and moves all others to the swap queue, which is
processed after the swap files. The queue is placed in swap mode so that
+ * FlowFiles added afterward are also processed after the swap files.
+ *
+ * This method MUST be called with the write lock held.
+ *
+ * @param oldestSwapLocation the location of the oldest recovered swap file
+ */
+ private void restoreActiveFlowFilesToSwapQueue(final String
oldestSwapLocation) {
+ swapMode = true;
+
+ final List<FlowFileRecord> swappedFlowFiles;
+ try {
+ swappedFlowFiles = swapManager.peek(oldestSwapLocation,
flowFileQueue).getFlowFiles();
+ } catch (final IOException ioe) {
Review Comment:
[Claude Opus 5.5] Yes, good point. Reading the whole swap file at startup is
new with this change. A corrupt file could throw an unchecked exception, which
would propagate out of `recoverSwappedFlowFiles()` and could prevent startup,
even though only ordering is at stake here. af957eec3cd catches `Exception` and
logs a warning.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/SwappablePriorityQueue.java:
##########
@@ -1267,6 +1267,10 @@ public SwapSummary recoverSwappedFlowFiles() {
incrementSwapQueueSize(swapFlowFileCount, swapByteCount,
swapLocations.size());
this.swapLocations.addAll(swapLocations);
+ if (!swapLocations.isEmpty()) {
+
restoreActiveFlowFilesToSwapQueue(swapLocations.iterator().next());
Review Comment:
[Claude Opus 5.5] Yes. `FlowFileSwapManager.recoverSwapLocations()`
documents that it returns the locations in the order that they should be
swapped back in, and `FileSystemSwapManager` sorts them oldest first.
`recoverSwappedFlowFiles()` keeps that order in a `LinkedHashSet`, and
`swapIn()` already relies on it by always taking `swapLocations.get(0)`.
af957eec3cd adds the `oldestSwapLocation` variable as suggested.
##########
nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/restart/FlowFileRestorationIT.java:
##########
@@ -85,6 +92,51 @@ public void
testDataInMissingQueueRestoredWhenQueueRestored() throws NiFiClientE
assertArrayEquals(flowFileContents, contentsAfterRestart);
}
+ @Test
+ public void testSwappedFlowFileOrderPreservedAfterRestart() throws
NiFiClientException, IOException, InterruptedException {
+ final int flowFileCount = 100;
+ ProcessorEntity generate =
getClientUtil().createProcessor(GENERATE_FLOWFILE);
+ generate = getClientUtil().updateProcessorProperties(generate,
Map.of("Batch Size", String.valueOf(flowFileCount), "sequence",
"${nextInt()}"));
+
+ ProcessorEntity verifyOrder =
getClientUtil().createProcessor("VerifySequentialOrder");
+ verifyOrder = getClientUtil().updateProcessorProperties(verifyOrder,
Map.of("Value to Consider", "${sequence}"));
+ final ConnectionEntity connection =
getClientUtil().updateConnectionPrioritizer(getClientUtil().createConnection(generate,
verifyOrder, SUCCESS), "FirstInFirstOutPrioritizer");
+
+ final ProcessorEntity terminate =
getClientUtil().createProcessor(TERMINATE_FLOWFILE);
+ final ConnectionEntity sequentialConnection =
getClientUtil().createConnection(verifyOrder, terminate, "sequential");
+ final ConnectionEntity failureConnection =
getClientUtil().createConnection(verifyOrder, terminate, "failure");
+
+ generate = getClientUtil().runProcessorOnce(generate);
Review Comment:
[Claude Opus 5.5] I don't think this can fail intermittently.
`FirstInFirstOutPrioritizer` compares the Last Queued Date and then the queue
date index. `StandardProcessSession` assigns that index from a static
`AtomicLong`, so it keeps increasing across all sessions. GenerateFlowFile
creates and transfers each FlowFile in turn, so FlowFiles that share a
millisecond are still ordered by index, which matches the order in which
`${nextInt()}` was evaluated. The two newer FlowFiles come from a separate,
later invocation, so they sort after the others for the same reason.
##########
nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/restart/FlowFileRestorationIT.java:
##########
@@ -85,6 +92,51 @@ public void
testDataInMissingQueueRestoredWhenQueueRestored() throws NiFiClientE
assertArrayEquals(flowFileContents, contentsAfterRestart);
}
+ @Test
+ public void testSwappedFlowFileOrderPreservedAfterRestart() throws
NiFiClientException, IOException, InterruptedException {
+ final int flowFileCount = 100;
+ ProcessorEntity generate =
getClientUtil().createProcessor(GENERATE_FLOWFILE);
+ generate = getClientUtil().updateProcessorProperties(generate,
Map.of("Batch Size", String.valueOf(flowFileCount), "sequence",
"${nextInt()}"));
+
+ ProcessorEntity verifyOrder =
getClientUtil().createProcessor("VerifySequentialOrder");
+ verifyOrder = getClientUtil().updateProcessorProperties(verifyOrder,
Map.of("Value to Consider", "${sequence}"));
+ final ConnectionEntity connection =
getClientUtil().updateConnectionPrioritizer(getClientUtil().createConnection(generate,
verifyOrder, SUCCESS), "FirstInFirstOutPrioritizer");
+
+ final ProcessorEntity terminate =
getClientUtil().createProcessor(TERMINATE_FLOWFILE);
+ final ConnectionEntity sequentialConnection =
getClientUtil().createConnection(verifyOrder, terminate, "sequential");
+ final ConnectionEntity failureConnection =
getClientUtil().createConnection(verifyOrder, terminate, "failure");
+
+ generate = getClientUtil().runProcessorOnce(generate);
+ getClientUtil().waitForStoppedProcessor(generate.getId());
+ waitForQueueCount(connection.getId(), flowFileCount);
+
+ // The oldest FlowFiles are held in memory and the rest are written to
swap files. Process two of the oldest FlowFiles and then queue
+ // two newer FlowFiles, which wait behind the swapped FlowFiles. All
FlowFiles held in memory then fit within the swap threshold when
+ // they are restored.
+ for (int i = 0; i < 2; i++) {
Review Comment:
[Claude Opus 5.5] Done in af957eec3cd. The test now declares
`newerFlowFileCount` once and uses it for the number of oldest FlowFiles
processed, the second Batch Size, the active queue size check, and the expected
total.
##########
nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/VerifySequentialOrder.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.processors.tests.system;
+
+import org.apache.nifi.annotation.behavior.TriggerSerially;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.List;
+import java.util.Set;
+
+@TriggerSerially
+@CapabilityDescription("Pulls one FlowFile at a time from the incoming queue
and routes it to 'sequential' if its Value to Consider is exactly one greater
than the "
+ + "Value to Consider of the previous FlowFile that this Processor
received. Otherwise, the FlowFile is routed to 'failure'. The first FlowFile
that the Processor "
+ + "receives is always routed to 'sequential'.")
Review Comment:
[Claude Opus 5.5] Changed to a text block in af957eec3cd.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
