Skip to content

fix: preserve duplicate same-timestamp elements in FlinkOrderedListState - #39783

Open
waterWang wants to merge 1 commit into
apache:masterfrom
waterWang:fix/flink-orderedliststate-duplicate-timestamps
Open

fix: preserve duplicate same-timestamp elements in FlinkOrderedListState#39783
waterWang wants to merge 1 commit into
apache:masterfrom
waterWang:fix/flink-orderedliststate-duplicate-timestamps

Conversation

@waterWang

Copy link
Copy Markdown

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):

// 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.

`readAsMap()` used `SortedMap.put()` keyed by timestamp, which silently
overwrites elements sharing the same timestamp. Changed to a multimap
structure (`SortedMap<Instant, List<TimestampedValue<T>>>`) using
`computeIfAbsent` to accumulate all values per timestamp bucket.

Fixes apache#39782

Signed-off-by: waterWang <waterwang@hermes.agent>
@github-actions

Copy link
Copy Markdown
Contributor

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant