[
https://issues.apache.org/jira/browse/NIFI-16468?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18131020#comment-18131020
]
ASF subversion and git services commented on NIFI-16468:
--------------------------------------------------------
Commit c8197c2402046f78e8ec31e70d0f363f4aadd42a in nifi's branch
refs/heads/main from Mark Payne
[ https://gitbox.apache.org/repos/asf?p=nifi.git;h=c8197c24020 ]
NIFI-16468: Return rolled-back FlowFiles to the active queue (#11815)
* NIFI-16468: Return rolled-back FlowFiles to the active queue
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.
> Session rollback can put FlowFile back on swap queue, changing queue order
> --------------------------------------------------------------------------
>
> Key: NIFI-16468
> URL: https://issues.apache.org/jira/browse/NIFI-16468
> Project: Apache NiFi
> Issue Type: Bug
> Components: Core Framework
> Reporter: Mark Payne
> Assignee: Mark Payne
> Priority: Major
> Time Spent: 1h 20m
> Remaining Estimate: 0h
>
> When a Processor pulls a FlowFile from a queue and then rolls back the
> session, the FlowFile is returned to its original queue using
> {{FlowFileQueue.put()}}. If that queue is in swap mode at the time,
> {{SwappablePriorityQueue}} adds the FlowFile to the swap queue instead of the
> active queue. The FlowFile was at the front of the active queue when it was
> polled, so it now ends up behind every other FlowFile in the active queue.
> This breaks ordering for any prioritizer, most noticeably the First In First
> Out prioritizer.
> This can be reproduced with a queue whose swap threshold is 10,000. Queue
> 10,001 FlowFiles, pull the first one, and roll back the session. The next
> pull returns the second FlowFile instead of the first.
> To address this, add {{void putToActive(FlowFileRecord)}} to
> {{FlowFileQueue}} and pass it through the queue partitions down to
> {{SwappablePriorityQueue}}, where it always adds to the active queue. Then
> use it from {{StandardProcessSession}} when rolling back.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
