markap14 commented on code in PR #11921: URL: https://github.com/apache/nifi/pull/11921#discussion_r4568546201
########## 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: [Claude Opus 5.5] A positive number validator would reject 0, and `${nextInt()}` starts at 0, so af957eec3cd uses `StandardValidators.LONG_VALIDATOR`, which accepts any whole number. Like the other numeric validators, it accepts values that contain Expression Language. `onTrigger` routes any evaluated value that is not a whole number to failure. ########## 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: [Claude Opus 5.5] Yes. The comparison uses a `QueuePrioritizer` built from the queue's configured prioritizers, which is the same comparator that orders the active queue, so nothing here is specific to FIFO. With `PriorityAttributePrioritizer`, restored FlowFiles with a higher priority than the highest priority FlowFile in the oldest swap file stay in the active queue. The rest wait in the swap queue until the swap files have been swapped in. For non-FIFO prioritizers, NiFi only orders the FlowFiles held in memory, because swap files are swapped in the order they were written. The one difference from before the restart is that a lower priority FlowFile that was in memory may now wait until after the swap files. That moves processing closer to priority order, not further from it. -- 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]
