szymonm opened a new pull request, #11918: URL: https://github.com/apache/nifi/pull/11918
# Summary [NIFI-16580](https://issues.apache.org/jira/browse/NIFI-16580) This PR currently adds **failing tests only, with no fix**; the fix will be added to this PR. They show that a FIFO-prioritized connection loses its order across a restart when its queue is in swap mode. Once the queue has swap files on disk and FlowFiles in the in-memory swap queue, FlowFiles enqueued later are polled after the restart *before* older FlowFiles from the swap file. The tests fail on purpose. They are **not** annotated with `@Disabled`, so they demonstrate the problem on current `main`. Reviewers may prefer to merge them with `@Disabled("NIFI-16580")` until a fix lands, or to merge them together with the fix. ## Mechanism (line references are to `main` at 0b218b02) 1. `FlowController.initializeFlow` calls `flowFileRepository.loadFlowFiles(queueProvider)` (`nifi-framework-core/.../controller/FlowController.java:1360`). It recovers each queue's swap files only afterwards, through `queue.recoverSwappedFlowFiles()` (`FlowController.java:1371`). 2. Before the restart, FlowFiles in the in-memory swap queue (not yet written to a swap file) are ordinary records in the write-ahead log. `SequentialAccessWriteAheadLog.recoverRecords` returns `recoveredRecords.values()` (`nifi-commons/nifi-write-ahead-log/.../wali/SequentialAccessWriteAheadLog.java:264`), and `HashMapSnapshot` backs that map with a `ConcurrentHashMap` (`HashMapSnapshot.java:50`). The records therefore come back in hash order of their ids, not in queue order. 3. `WriteAheadFlowFileRepository.loadFlowFiles` puts the records into the queue one by one (`WriteAheadFlowFileRepository.java:953`). `SwappablePriorityQueue.put` fills the active queue up to `swapThreshold` and sends the rest to the swap queue (`SwappablePriorityQueue.java:533`). Old active FlowFiles and newer swap-queue FlowFiles compete for active slots in hash order, so newer FlowFiles can take active slots. 4. `recoverSwappedFlowFiles` appends the recovered swap locations afterwards (`SwappablePriorityQueue.java:1196`, `:1269`). The swap files are swapped in only once the active queue drains, and `migrateSwapToActive` does nothing while the active queue is non-empty (`SwappablePriorityQueue.java:340`, `:353`). By then the newer FlowFiles in the active queue have already been polled ahead of the older swapped-out ones. The bug reproduces after a graceful shutdown as well as a crash: shutdown neither swaps out the in-memory swap queue nor checkpoints the FlowFile repository. It needs a checkpoint written after the swap-out; recovery then reads a small map whose hash order differs from id order. When recovery instead replays a long journal into a large map, small sequential ids happen to come back in ascending order. ## Tests ### System test - `SwapFifoOrderAcrossRestartIT` (nifi-system-tests): a real NiFi instance with `nifi.queue.swap.threshold=64` and a 1 second FlowFile repository checkpoint interval. It queues 4,159 FlowFiles with an increasing index behind a stopped processor, so 0-63 are active, 64-4,095 are in 63 swap files and 4,096-4,158 stay in memory. It waits for a checkpoint newer than the last swap file, stops and starts NiFi, and checks the queue order. It fails with `FIFO order not preserved after restart: queue position 32 holds FlowFile with index 4096 but only indexes 0..63 were enqueued first`. The checkpoint holds 127 records, the recovered map has 256 buckets, and id 4,096+i shares a bucket with id i, so this does not depend on the starting id. ### Framework tests These use a swap threshold of 100 and enqueue 250 FlowFiles in FIFO order: - indexes 0-99 are in the active queue - indexes 100-199 are in one swap file - indexes 200-249 are in the in-memory swap queue, persisted in the WAL After a restart, every test drains the queue and asserts that FlowFiles come out in enqueue order. - `TestWriteAheadFlowFileRepository#testFifoOrderPreservedAcrossRestartWithSwappedFlowFiles`: real `WriteAheadFlowFileRepository` and `StandardFlowFileQueue`, with an in-memory mock swap manager. It also fixes an NPE in the test's `MockFlowFileSwapManager.swapIn`. - `TestWriteAheadFlowFileRepository#testFifoOrderPreservedAcrossRestartWithFileSystemSwapManager`: the real `FileSystemSwapManager`, with on-disk swap files reused across the restart. The queue swaps out on its own when it crosses the threshold. - `FlowControllerTest#testFifoOrderPreservedAcrossRestartWithSwappedFlowFiles`: two standalone `FlowController` instances share the same repository directories, with `nifi.queue.swap.threshold=100`. The test builds a funnel-to-funnel connection with fixed ids, persists FlowFiles through the controller's FlowFile repository, and shuts down. It then rebuilds the same connection and restarts through the real `FlowController.initializeFlow()`. All three framework tests fail with: ``` Newer FlowFile index 220 from the in-memory swap queue was polled at position 70, before older FlowFile index 100 from the swap file at position 100 ``` ### Notes - **Determinism:** the reordering depends on the hash order of the FlowFile ids. The tests use ids starting at `2^20 - 220`, so the range crosses a power-of-two boundary and the hash iteration order differs from id order the same way on every run. With small sequential ids, iteration order can match id order and hide the problem. - **Prioritizer:** `FirstInFirstOutPrioritizer` lives in `nifi-standard-prioritizers` and is not on the `nifi-framework-core` test classpath. The tests inline an equivalent comparator ordered by last queue date, then queue date index. - **Queue ids:** `FileSystemSwapManager` requires queue ids that are UUIDs, so the tests use fixed UUIDs to re-create the same queue after the restart. ## How to run ``` ./mvnw -pl nifi-framework-bundle/nifi-framework/nifi-framework-core test \ -Dtest='TestWriteAheadFlowFileRepository#testFifoOrder*,FlowControllerTest#testFifoOrder*' \ -Dsurefire.failIfNoSpecifiedTests=false ./mvnw -pl nifi-system-tests/nifi-system-test-suite verify -Pintegration-tests \ -Dit.test=SwapFifoOrderAcrossRestartIT -Dtest=none -Dsurefire.failIfNoSpecifiedTests=false ``` The system test needs `nifi-assembly` to be built first. ## Possible fix directions (not part of this PR) - Sort the records restored for each queue by the queue's comparator before putting them, or at least by queue date and then id, instead of relying on the WAL's hash order. - Recover swap locations before loading the WAL records, so `put` sees the queue as already in swap mode. Records restored from the WAL that are newer than the swap files should then go to the swap queue rather than the active queue. - Either way, restored records that belong behind the swap files (newer than the oldest swapped FlowFile) should not land in the active queue ahead of them. # Tracking ### 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 ### Build - [ ] Build completed using `./mvnw clean install -P contrib-check` - [ ] JDK 21 - [ ] JDK 25 `checkstyle:check pmd:check` passes for `nifi-framework-core` and `nifi-system-test-suite`. The four new tests fail on purpose. The other tests in the modified classes still pass. ### Licensing - [x] No new dependencies ### Documentation - [x] No documentation changes -- 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]
