[
https://issues.apache.org/jira/browse/FLINK-41256?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Piotr Nowojski updated FLINK-41256:
-----------------------------------
Affects Version/s: 2.4.0
> Operators changing the current key out of order silently lose state in BATCH
> execution mode
> -------------------------------------------------------------------------------------------
>
> Key: FLINK-41256
> URL: https://issues.apache.org/jira/browse/FLINK-41256
> Project: Flink
> Issue Type: Bug
> Components: API / DataStream, Runtime / Task
> Affects Versions: 2.3.0, 2.2.1, 2.1.3, 2.4.0
> Reporter: Piotr Nowojski
> Assignee: Piotr Nowojski
> Priority: Critical
> Labels: pull-request-available
> Fix For: 2.5.0
>
>
> In BATCH execution mode, keyed operators run on
> BatchExecutionKeyedStateBackend with key-sorted inputs. The backend keeps the
> state of only one key at a time: whenever the current key changes, it fires
> all timers of the outgoing key and clears all of its state.
> This assumes the current key only ever advances with the sorted input.
> Nothing enforces that: an operator can call setCurrentKey(...) itself
> (AbstractStreamOperator#setCurrentKey is public and reachable from custom
> operators added via DataStream#transform). An operator that buffers records
> of several keys and later processes them key by key goes back to keys that
> the input has already left. That is correct in STREAMING, but in BATCH it
> silently:
> * reads empty state for the revisited key, as its state was already dropped,
> * fires the outgoing key's timers early, before all of that key's records
> were processed.
> The job finishes successfully and produces wrong results.
> h3. Reproduction
> A minimal custom keyed operator: it buffers 2 records of any keys, then sums
> each buffered record into a ValueState after setCurrentKey(record key), and
> also flushes on endInput.
> Input (keyed by f0): (a,1), (a,1), (a,1), (b,1)
> * STREAMING: a=3, b=1 (correct)
> * BATCH: a=1, b=1
> In BATCH, a1 and a2 are flushed into a's state (a=2) and a3 stays buffered.
> When b1 arrives, the backend switches to b and clears a's state. The flush of
> [a3, b1] then switches back to a, reads empty state and computes a=1.
> Python operators already work around this in BATCH:
> AbstractPythonFunctionOperator#setCurrentKey flushes its bundle before the
> key changes.
> h3. Proposed fix
> Make the problem fail loudly and give operators a hook to flush before the
> key changes:
> * Add KeyContext#moveToNextKey and KeyedStateBackend#moveToNextKey, which
> must never go back to a previous key. They default to setCurrentKey, so
> streaming backends are unaffected. Add KeyedStateBackend#requiresKeyOrder,
> true only for the batch backend.
> * The keyed input path of AbstractStreamOperator(V2) uses moveToNextKey when
> the backend requires key order.
> * BatchExecutionKeyedStateBackend#setCurrentKey throws for any key other than
> the current one.
> * Operators that buffer across keys override moveToNextKey to flush first.
> The Python operators and the State Processor API bootstrap wrapper hook into
> moveToNextKey instead of setCurrentKey.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
