markap14 opened a new pull request, #11921: URL: https://github.com/apache/nifi/pull/11921
# Summary [NIFI-16580](https://issues.apache.org/jira/browse/NIFI-16580) When NiFi restarts, a connection's FlowFiles are restored in two steps. First, the FlowFiles that were held in memory are loaded from the FlowFile Repository. Then the connection's swap files are recovered. The FlowFiles from the first step fill the active queue no matter when they were queued. Because a swap file is only swapped back in once the active queue is empty, FlowFiles that were queued after the swapped FlowFiles could be processed before them. Last Queued Date doesn't prevent this: the prioritizer only orders the FlowFiles that are in the active queue, and the swapped FlowFiles aren't in it yet. This change updates `SwappablePriorityQueue.recoverSwappedFlowFiles()`. When swap files are recovered, it reads the oldest swap file and finds its highest priority FlowFile. Restored FlowFiles that come before that FlowFile stay in the active queue. All others are moved to the swap queue, which is processed after the swap files. The queue is also placed in swap mode so that FlowFiles added afterward wait behind the swap files too. This is an alternative to #11918 that keeps the change limited to `SwappablePriorityQueue`. ### Tests - `TestSwappablePriorityQueue.testRecoverSwappedFlowFilesPreservesOrder` restores FlowFiles from both before and after a swap file, recovers the swap file, and verifies that every FlowFile is polled in order. - `FlowFileRestorationIT.testSwappedFlowFileOrderPreservedAfterRestart` is a new system test. With a swap threshold of 20, it queues 100 FlowFiles, processes the two oldest, and queues two newer FlowFiles behind the swapped FlowFiles. After a restart, it processes everything and checks that no FlowFile was processed out of order. - `VerifySequentialOrder` is a new system test processor used by that test. It routes each FlowFile to `sequential` if its `Value to Consider` is one greater than the previous FlowFile's value and to `failure` otherwise. Both new tests fail without the change to `SwappablePriorityQueue` and pass with it. # Tracking Please complete the following tracking steps prior to pull request creation. ### Issue Tracking - [x] [Apache NiFi Jira](https://issues.apache.org/jira/browse/NIFI) issue created ### Pull Request Tracking - [x] Pull Request title starts with Apache NiFi Jira issue number, such as `NIFI-00000` - [x] Pull Request commit message starts with Apache NiFi Jira issue number, as such `NIFI-00000` - [x] Pull request contains [commits signed](https://docs.github.com/en/authentication/managing-commit-signature-verification/signing-commits) with a registered key indicating `Verified` status ### Pull Request Formatting - [x] Pull Request based on current revision of the `main` branch - [x] Pull Request refers to a feature branch with one commit containing changes # Verification Please indicate the verification steps performed prior to pull request creation. ### Build - [ ] Build completed using `./mvnw clean install -P contrib-check` - [ ] JDK 21 - [ ] JDK 25 ### Licensing - [ ] New dependencies are compatible with the [Apache License 2.0](https://apache.org/licenses/LICENSE-2.0) according to the [License Policy](https://www.apache.org/legal/resolved.html) - [ ] New dependencies are documented in applicable `LICENSE` and `NOTICE` files ### Documentation - [ ] Documentation formatting appears as expected in rendered files -- 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]
