exceptionfactory commented on code in PR #11921:
URL: https://github.com/apache/nifi/pull/11921#discussion_r4568326305
##########
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:
Should `RuntimeException` also be caught and logged here?
##########
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:
It looks like `swapMode` should only be changed after a successful
`swapManager.peek()`, unless there are other reasons to make the switch
regardless of status.
##########
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:
This can be changed to a multiline string instead of using concatenation
##########
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:
The number `2` here and below would be helpful to declare once and reuse to
indicate the number of newer FlowFiles expected
##########
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:
Is it safe to presume that the `swapLocations` are always ordered?
For clarity, it seems useful to declare `final String oldestSwapLocation =
swapLocations.iterator().next()` to highlight that expectation
##########
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) {
+ logger.warn("Failed to read Swap File {} for {}; FlowFiles
restored from the FlowFile Repository may be processed before older swapped
FlowFiles",
+ oldestSwapLocation, this, ioe);
+ return;
+ }
+
+ if (swappedFlowFiles.isEmpty()) {
+ return;
+ }
+
+ final QueuePrioritizer prioritizer = new
QueuePrioritizer(getPriorities());
+ final FlowFileRecord firstSwappedFlowFile =
Collections.min(swappedFlowFiles, prioritizer);
Review Comment:
Does this approach also work for non-FIFO Prioritizers like the Property
Attribute Prioritizer?
##########
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'.")
+public class VerifySequentialOrder extends AbstractProcessor {
+
+ static final PropertyDescriptor VALUE_TO_CONSIDER = new
PropertyDescriptor.Builder()
+ .name("Value to Consider")
+ .description("The whole number to compare against the Value to
Consider of the previous FlowFile")
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
Review Comment:
Should this use some kind of positive number validator?
##########
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:
Is there a potential for intermittent failure in the test if multiple
FlowFiles have the same millisecond timestamp? It seems unlikely, but possible
to avoid with multiple invocations of the Processor.
--
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]
