waterWang opened a new pull request, #39783:
URL: https://github.com/apache/beam/pull/39783
**Issue:** #39782
**Problem:** `FlinkOrderedListState.readAsMap()` builds a
`SortedMap<Instant, TimestampedValue<T>>` using `Map.put()` keyed by timestamp.
When multiple elements share the same timestamp, the second element overwrites
the first — causing silent data loss when reading the state.
**Fix:** Changed to a multimap structure (`SortedMap<Instant,
List<TimestampedValue<T>>>`). Uses `Map.computeIfAbsent()` to accumulate all
values per timestamp bucket, preserving every element. Updated `read()`,
`readRange()`, and `clearRange()` to flatten the multimap values.
**Root cause (code):**
```java
// Before (bug): duplicates lost
SortedMap<Instant, TimestampedValue<T>> sortedMap = Maps.newTreeMap();
for (TimestampedValue<T> value : listValues) {
sortedMap.put(value.getTimestamp(), value); // overwrites same-ts
elements
}
// After: all values preserved
SortedMap<Instant, List<TimestampedValue<T>>> sortedMap = Maps.newTreeMap();
for (TimestampedValue<T> value : listValues) {
sortedMap.computeIfAbsent(value.getTimestamp(), k -> new
ArrayList<>()).add(value);
}
```
**Testing:** `FlinkStateInternalsTest` — existing ordered-list tests cover
ordering semantics; they pass unchanged since the multimap flattens back to the
same sorted order. The scoped PR change is behavior-preserving for unique
timestamps and fixes duplicate-timestamp data loss.
**Docs / changelog:** This is a bug fix; no user-facing API change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
