Skip to content

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

Open

Description

@vemuriraghu

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<Long, List<T>>) 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
  • Component: Flink Runner
  • Component: Prism Runner
  • Component: Twister2 Runner
  • Component: Hazelcast Jet Runner
  • Component: Google Cloud Dataflow Runner

Activity

  1. je-ik commented on Aug 17, 2026

    @je-ik
    Contributor

    The problem is that FlinkOrderedListState.readAsMap() returns Map<Instant, TimestampedValue<T>>, which prevents multiple records with the same Instant. This would have to be changed to MultiMap , PriorityQueue or similar approach. Also there is missing @ValidatesRunner suite that will cover this behavior.

    Overall (due to missing equivalent in Flink state primitives) the implementation is quite inefficient (mostly if the list grows larger), as it sorts the complete list on every read() or readRange().

  2. je-ik commented on Sep 8, 2026

    @je-ik
    Contributor

    Should the duplicate PR be closed? #39795

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

Metadata

Metadata

Assignees

No one assigned

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions