pnowojski opened a new pull request, #29658:
URL: https://github.com/apache/flink/pull/29658
## What is the purpose of the change
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 the
outgoing key's timers and clears its state.
Nothing enforced that the current key only advances with the sorted input.
An operator can call `setCurrentKey(...)` itself, which is public on
`AbstractStreamOperator` and reachable from custom operators added via
`DataStream#transform`. For example, an operator that buffers records of
several keys and later processes them key by key goes back to keys the input
has already left. That works in STREAMING, but in BATCH it silently reads empty
state for the revisited key and fires timers early. The job succeeds with wrong
results (FLINK-41256).
This pull request makes such operators fail loudly in BATCH, and gives
operators a hook to flush their buffers before the key changes.
## Brief change log
- Add `KeyContext#moveToNextKey` and `KeyedStateBackend#moveToNextKey`:
move to the next key of a key-sorted input, never back to a previous one. They
default to `setCurrentKey`, so streaming backends are unaffected.
- Add `KeyedStateBackend#requiresKeyOrder`, `true` only for
`BatchExecutionKeyedStateBackend`.
- The keyed input path of `AbstractStreamOperator(V2)` uses
`moveToNextKey` when the backend requires key order. Otherwise it keeps calling
`setCurrentKey`, so existing overrides still see every key.
- `BatchExecutionKeyedStateBackend#setCurrentKey` now throws for any key
other than the current one.
- The Python operators and the State Processor API bootstrap wrapper,
which hooked into `setCurrentKey` for BATCH key changes, now hook into
`moveToNextKey` too.
- Async state context switches (`setCurrentKeyAndKeyGroup`) are not
checked yet.
## Verifying this change
This change added tests and can be verified as follows:
-
`DataStreamBatchExecutionITCase#batchOperatorGoingBackToAPreviousKeyFails`: a
custom operator that buffers records across keys and processes them via
`setCurrentKey` produces correct sums in STREAMING and fails in BATCH. Without
the fix, BATCH silently returned `a=1` instead of `a=3`.
- `DataStreamBatchExecutionITCase#batchOperatorFlushingBeforeTheNextKey`:
the same operator, flushing its buffer in `moveToNextKey`, produces correct
sums in both modes.
- `BatchExecutionKeyOrderTest`: the input path moves through keys, and an
operator switching back to a previous key fails.
- Existing batch state backend and batch timer service tests now switch
keys via `moveToNextKey`.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: yes. `AbstractStreamOperator` (`@PublicEvolving`) gets a
new `moveToNextKey` method. `KeyContext` and `KeyedStateBackend` get new
default methods. In BATCH, operators that called `setCurrentKey` with a key
other than the current one now fail instead of silently producing wrong results.
- The serializers: no
- The runtime per-record code paths (performance sensitive): yes. The
keyed input path now goes through `moveToNextKey`, which costs one extra check
per record.
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? JavaDocs
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Claude Opus 5.5)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
