Here is the bug - https://github.com/apache/beam/issues/39782
On Monday, August 17, 2026 at 01:02:31 AM PDT, Jan Lukavský
<[email protected]> wrote:
Hi,
confirming this looks like a bug.
FlinkStateInternals.FlinkOrderedListState.readAsMap keys elements by Instant,
which is not guaranteed to be unique. Can you please fill a github issue for
this?
Jan
On 8/17/26 08:29, Raghu Vemuri via user wrote:
## 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)
