What happened?
Component
runner-flink
What happened?
When executing a pipeline using the FlinkRunner (specifically observed using the HashMapStateBackend / Java heap state configuration), adding multiple distinct elements with the exact same timestamp to an OrderedListState results in critical data loss. The last element written completely overwrites any previous elements that shared that specific millisecond timestamp bucket.
According to Beam's core architectural contract, OrderedListState should preserve and sort all appended values. It does not impose an invariant rule requiring timestamps to be globally unique.
The underlying flaw stems from how FlinkStateInternals maps OrderedListState onto Flink's native primitive structures. The runner sets the key of Flink's internal tracking storage structure to the Long timestamp value itself. During back-to-back writes in a single execution bundle, the internal write-buffer performs a flat replacement (equivalent to a map put action) instead of appending the element to a sub-collection (like a List) associated with that timestamp bucket.
Expected Behavior
OrderedListState should treat the timestamp solely as a sorting key index. If multiple elements share a timestamp, the runner's underlying serialization layout must wrap elements in an appendable multi-value collection (e.g., mapping to a Flink MapState<Long, List<T>>) so that subsequent writes do not execute a flat destructive overwrite.
Actual Behavior
The second write ("Ball") silently destroys and overwrites the initial write ("Apple") at the timestamp boundary level inside the Java heap state backend.
Environment
Apache Beam Version: 2.61.0
Flink Version: 1.19.1
Runner: FlinkRunner
State Backend: HashMapStateBackend (Java Heap)
Issue Priority
Priority: 1 (data loss / total loss of function)
Issue Components
What happened?
Component
runner-flink
What happened?
When executing a pipeline using the
FlinkRunner(specifically observed using theHashMapStateBackend/ Java heap state configuration), adding multiple distinct elements with the exact same timestamp to anOrderedListStateresults in critical data loss. The last element written completely overwrites any previous elements that shared that specific millisecond timestamp bucket.According to Beam's core architectural contract,
OrderedListStateshould preserve and sort all appended values. It does not impose an invariant rule requiring timestamps to be globally unique.The underlying flaw stems from how
FlinkStateInternalsmapsOrderedListStateonto Flink's native primitive structures. The runner sets the key of Flink's internal tracking storage structure to theLongtimestamp value itself. During back-to-back writes in a single execution bundle, the internal write-buffer performs a flat replacement (equivalent to a mapputaction) instead of appending the element to a sub-collection (like aList) associated with that timestamp bucket.Expected Behavior
OrderedListStateshould treat the timestamp solely as a sorting key index. If multiple elements share a timestamp, the runner's underlying serialization layout must wrap elements in an appendable multi-value collection (e.g., mapping to a FlinkMapState<Long, List<T>>) so that subsequent writes do not execute a flat destructive overwrite.Actual Behavior
The second write (
"Ball") silently destroys and overwrites the initial write ("Apple") at the timestamp boundary level inside the Java heap state backend.Environment
Apache Beam Version: 2.61.0
Flink Version: 1.19.1
Runner: FlinkRunner
State Backend:
HashMapStateBackend(Java Heap)Issue Priority
Priority: 1 (data loss / total loss of function)
Issue Components