Szymon Matejczyk created NIFI-16580:
---------------------------------------
Summary: FlowFile order not preserved across restart for a
connection with swapped-out FlowFiles
Key: NIFI-16580
URL: https://issues.apache.org/jira/browse/NIFI-16580
Project: Apache NiFi
Issue Type: Bug
Components: Core Framework
Affects Versions: 2.12.0
Reporter: Szymon Matejczyk
h3. Bug
When NiFi restarts while a connection is in swap mode, the FlowFiles in that
connection are not restored in their original order. FlowFiles that were held
in memory (the active queue and the in-memory swap queue) are restored first
and in arbitrary order, and they all come out ahead of the older FlowFiles that
were swapped out to disk. A consumer can therefore receive newer FlowFiles
before older ones, even with FirstInFirstOutPrioritizer configured.
The older, swapped-out FlowFiles are only swapped back in once the active queue
is completely empty. On a busy or backpressured connection this can take a long
time, so the reordering is not limited to a few FlowFiles.
Flows that rely on FIFO ordering within a connection (for example, a marker
FlowFile that must be processed before the FlowFiles that follow it) can
process data out of order after a restart.
This happens after both a graceful shutdown and a sudden one (for example kill
-9). A graceful shutdown neither swaps out the in-memory swap queue nor writes
a FlowFile repository checkpoint, so the on-disk state that recovery starts
from has the same shape in both cases.
h3. Mechanism
FlowController.initializeFlow calls FlowFileRepository.loadFlowFiles(...)
before FlowFileQueue.purgeSwapFiles() and recoverSwappedFlowFiles(). The
FlowFiles from the repository are loaded into the queues before the queues know
about their swap files.
WriteAheadFlowFileRepository.loadFlowFiles iterates the records returned by the
write-ahead log and calls FlowFileQueue.put(...) for each one.
SequentialAccessWriteAheadLog.recoverRecords() returns the values of a hash map
keyed by FlowFile id (HashMapSnapshot), so the records are put in hash order,
not enqueue order. At this point, records that were in the active queue and
records that were in the in-memory swap queue cannot be told apart.
SwappablePriorityQueue.put places the first records, up to the swap threshold,
in the active queue and the rest in the in-memory swap queue. Because the order
is arbitrary, the newest FlowFiles can end up in the active queue.
recoverSwappedFlowFiles() then appends the recovered swap files after these
records.
SwappablePriorityQueue.migrateSwapToActive does nothing while the active queue
is non-empty. Once the active queue is empty, it swaps in the swap files in
order, and moves the in-memory swap queue only after all swap files are
consumed.
The prioritizer only orders FlowFiles within the active queue, so it cannot
correct this: the older FlowFiles are not in the active queue to be compared.
h3. When it happens Three conditions must hold at restart:
* The connection has swapped-out FlowFiles and also FlowFiles in the in-memory
swap queue. Swap files are written in batches of min(10,000, swap threshold),
so up to one batch minus one FlowFile stays in memory.
* The FlowFile repository was checkpointed after the swap-out. Recovery then
reads a small checkpoint into a small hash map. Without a recent checkpoint,
recovery replays the journal into a map sized for every FlowFile ever created
since the last checkpoint, and for small sequential ids that map happens to
iterate in ascending id order, which hides the bug.
* Some in-memory FlowFile ids fall into the same hash buckets as the active
FlowFile ids, so they are iterated before some of the older active FlowFiles.
With a map of capacity C, id i and id i + k*C share a bucket.
With the default checkpoint interval, the second condition is normally met on a
running instance, so whether the bug shows up depends mainly on the FlowFile
ids.
h3. h3. Steps to reproduce
Start NiFi with nifi.queue.swap.threshold=64 and a short
nifi.flowfile.repository.checkpoint.interval (for example 1 sec).
Build a flow: a source processor -> connection with FirstInFirstOutPrioritizer
-> stopped destination processor.
Generate 4,159 FlowFiles with an increasing index attribute. FlowFiles 0-63
stay in the active queue, 64-4,095 are written to 63 swap files, and
4,096-4,158 stay in the in-memory swap queue.
Wait until the FlowFile repository has been checkpointed after the last swap
file was written.
Stop and start NiFi.
List or poll the connection.
Expected: FlowFiles come out in enqueue order, 0 to 4,158. Actual: newer
FlowFiles come out first. The checkpoint holds 127 records, the recovered map
has 256 buckets, and id 4,096+i shares a bucket with id i. The active queue is
therefore filled with 0-31 and 4,096-4,127, and queue position 32 holds the
FlowFile with index 4,096 instead of 32.
h3. Pull request A pull request will follow. For now it contains only failing
tests that reproduce the bug; the fix will be added to the same pull request.
* SwapFifoOrderAcrossRestartIT (nifi-system-tests): the scenario above with a
real NiFi instance, a real stop and start, and an assertion on queue order.
* TestWriteAheadFlowFileRepository: restart of the FlowFile repository with a
mock swap manager, and with the real FileSystemSwapManager.
* FlowControllerTest: restart of a real FlowController over the same
repository and swap directories.
The tests choose FlowFile counts and ids so that the hash order, and therefore
the failure, is deterministic. They depend on the bucket layout of
java.util.HashMap.
h3. RelatedÂ
NIFI-16468 (fixed in 2.13.0) fixed rolled-back FlowFiles being re-queued behind
swapped FlowFiles. That fix does not change restart recovery, and this bug is
still present on main.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
