Repository navigation
Conversation
Session rollback used FlowFileQueue.put(), which places a FlowFile on the swap queue when the queue is swapping. That puts it behind FlowFiles that were already in the active queue. Co-authored-by: Cursor <cursoragent@cursor.com>
| final long expirationEpochMillis = System.currentTimeMillis() + context.getConnectable().getPenalizationPeriod(TimeUnit.MILLISECONDS); | ||
| final FlowFileRecord newFile = new StandardFlowFileRecord.Builder().fromFlowFile(record.getOriginal()).penaltyExpirationTime(expirationEpochMillis).build(); | ||
| originalQueue.put(newFile); | ||
| originalQueue.putToActive(newFile); |
There was a problem hiding this comment.
StatelessFlowTask (lines 397 and 404) also puts polled FlowFiles back with put() when the stateless group fails. Should it use putToActive() too, so the same reordering does not happen there?
There was a problem hiding this comment.
No need, in Stateless, we use a StatelessFlowFileQueue which does not support swapping. So the put vs putToActive is irrelevant there.
| * ahead of any FlowFiles that are swapped out, and placing it back onto the active queue preserves that ordering. | ||
| * | ||
| * @param file the FlowFile to place onto the active queue | ||
| */ |
There was a problem hiding this comment.
Should putToActive have a default implementation that calls put(file), so that FlowFileQueue implementations outside this repo keep compiling?
There was a problem hiding this comment.
No, there's no way to configure instantiation of FlowFileQueue objects outside of what exists in this repo.
There was a problem hiding this comment.
I meant that the source queue for the Stateless PG comes from port.getIncomingConnections() in the parent PG, not from the stateless engine. Can that queue swap meaning that lines 397 and 404 have the same reordering issue?
2 participants
Summary
NIFI-16468
When a Processor pulls a FlowFile and then rolls the session back, the FlowFile was returned with
FlowFileQueue.put(). If the queue is swapping, that method places the FlowFile on the swap queue, behind every FlowFile still in the active queue.putToActivealways places it on the active queue, which is where it was when it was pulled.Tracking
Please complete the following tracking steps prior to pull request creation.
Issue Tracking
Pull Request Tracking
NIFI-16468NIFI-16468VerifiedstatusPull Request Formatting
mainbranchVerification
StandardProcessSessionIT.testRollbackReturnsFlowFileToFrontOfSwappingQueuequeues 10,001 FlowFiles, pulls the first, rolls the session back, and checks that the next pull returns that same FlowFile.TestSwappablePriorityQueuecovers the same return path on the queue itself.Build
./mvnw clean install -P contrib-checkLicensing
LICENSEandNOTICEfilesDocumentation