[waterWang]

Fix FlinkOrderedListState data loss when multiple elements with the same timestamp are added to an OrderedListState.

Root cause: The readAsMap() method builds a TreeMap<Instant, TimestampedValue<T>> keyed by the element's timestamp. When a timestamp is shared by multiple elements, Map.put() silently overwrites the previous entry, destroying data.

Fix: Change the internal structure to SortedMap<Instant, List<TimestampedValue<T>>>, using computeIfAbsent to append same-timestamp elements to a list instead of overwriting. All callers (read(), readRange(), clearRange()) are updated to flatten the list via Iterables.concat().

Fixes #39782

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[je-ik]

Thanks for the fix!

LGTM in general, but it would be good to cover this change in flink-specific tests. Even better would be if we could cover this in ParDoTest suite (next to testOrderedListStateUnbounded). This might reveal multiple runners where similar fix should be applied (in order not to block this PR, I'd suggest doing both - add flink-specific test and a different PR that will change the ParDoTest that is applied to multiple runners).

Also the PR fails spotless check - it can be fixed by ./gradlew spotlessApply

[shunping]

@waterWang: any update regarding the last review?

[shunping]

waiting on author