Piotr Nowojski created FLINK-41256:
--------------------------------------
Summary: 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.1.3, 2.2.1, 2.3.0
Reporter: Piotr Nowojski
Assignee: Piotr Nowojski
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)
