## Componentrunner-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 BehaviorThe 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)
