apache / apache/beam

[Bug]: Java FlinkRunner OrderedListState silently overwrites duplicate elements sharing same timestamps

Open
#39,782 2 comments 0 reactions 0 assignees View on GitHub
bug flink P1
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

### What happened?

## 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>`) 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)

### Issue Priority

Priority: 1 (data loss / total loss of function)

### Issue Components

- [ ] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [x] Component: Flink Runner
- [ ] Component: Prism Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner

Contributor guide

Open the contributing guide

Research direction

Start by reading FlinkStateInternals and tracing how OrderedListState writes are represented with the Flink HashMapStateBackend. Reproduce the two same-timestamp writes using the reported Apple and Ball case, then verify that both values are retained and sorted rather than one replacing the other.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.