twalthr opened a new pull request, #29657:
URL: https://github.com/apache/flink/pull/29657
## What is the purpose of the change
`ProcessTableFunction.Context#clearAllTimers()` (and `clearAll()`) did not
scope the removal of unnamed timers to the current partition key.
`WritableInternalTimeContext#clearUnnamedTimers` iterated the unnamed timers
via `InternalTimerService#forEachEventTimeTimer`. That iteration covers the
timers of **all keys** in the subtask and sets the key context to the key of
each visited timer without restoring it. As a result:
1. Timestamps of other keys' timers were collected.
2. The deletes ran under the key of the last visited timer instead of the
current key. Timers of the current key survived and fired later, and timers of
other keys with the same timestamp were silently deleted.
3. The operator's key context was left on an arbitrary key. Later calls in
the same `eval()`/`onTimer()` ran under the wrong key: the clearing of named
timers in `clearAllTimers()`, `clearAllState()` in `clearAll()`, and the
write-back of state entries.
Named timers are tracked in keyed map state, so the timer lookup itself was
key-scoped. They were only affected indirectly, through the corrupted key
context described in 3.
## Brief change log
- `WritableInternalTimeContext#clearUnnamedTimers` only collects the
timers of the current key and restores the key context afterwards.
- `AbstractProcessTableOperator` passes the `KeyedStateBackend` to
`WritableInternalTimeContext`.
- The key comparison relies on value-based key equality, which is only
guaranteed for `BinaryRowData`. The key selectors of the PTF node always
produce `BinaryRowData`, and the heap state backend already relies on key
`equals` for its own lookups. To make this invariant explicit,
`AbstractProcessTableOperator#setCurrentKey` (through which every key enters
the operator) now throws an `IllegalStateException` for any other key type.
- No new state is introduced, so there is no change to the state layout or
checkpoint compatibility, and the timer registration path is unchanged.
Clearing still iterates all unnamed timers of the subtask, but that only
happens when `clearAllTimers()` is called. A per-key index of unnamed timers
could make this cheaper but would change the state layout, so it is left for a
follow-up.
## Verifying this change
Please make sure both new and modified tests in this PR follow [the
conventions for tests defined in our code quality
guide](https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#7-testing).
This change added tests and can be verified as follows:
- Added `PROCESS_CLEAR_ALL_UNNAMED_TIMERS_PER_KEY` to
`ProcessTableFunctionSemanticTests`: three keys register an unnamed timer for
the same time, two of them call `clearAllTimers()`, only the untouched key's
timer fires, and the cleared keys keep consistent state afterwards. The test
fails without the fix and passes with it.
- Added `ProcessSetTableOperatorTest#testNonBinaryKey` which verifies that
non-binary keys are rejected.
- Additionally verified manually (not part of the PR) that `equals` of the
keys agrees with equality of the serialized keys, for 15 partition key types on
the heap and RocksDB state backends (with both timer service factories) and
after a serializer round trip.
- Ran spotless, checkstyle, and the ArchUnit tests locally.
## 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)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): yes (a single
`instanceof` check when the key context is set)
- 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? not applicable
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Claude Sonnet 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]
