Grzegorz Liter created FLINK-40442:
--------------------------------------
Summary: State Processor API re-derives a key's key-group from
hashCode() instead of reusing the one it read the key from — breaks reading old
state after any hashCode() change
Key: FLINK-40442
URL: https://issues.apache.org/jira/browse/FLINK-40442
Project: Flink
Issue Type: Improvement
Components: API / State Processor
Reporter: Grzegorz Liter
When reading keyed state with SavepointReader.readKeyedState(), each key is
discovered by iterating storage per key-group — the backend already knows which
key-group a key lives under while doing this. But KeyedStateInputFormat throws
that away and calls setCurrentKey(key), which recomputes the key-group from
key.hashCode() instead of reusing the one it was just found under. If
hashCode()/equals() for that key type ever changed since the savepoint was
written, the recomputed key-group can disagree with where the key actually
lives, and you get:
{code}
Key group X is not in KeyGroupRange\{...}
{code}
This makes affected state permanently unreadable via the State Processor API —
even though it's fully intact and a live streaming job restores it fine (live
restore only computes a key-group once, when a new record first arrives; it
never re-derives one for already-stored keys).
This is exactly the failure mode reported in AVRO-4334 (Avro's Utf8.hashCode()
changing undocumented between 1.12.0 and 1.12.1). Fixing whatever caused a
hashCode() to drift doesn't help with data already written under the old hash —
this bug is what stands between that state and being repairable at all.
Proposed fix
AbstractKeyedStateBackend already has setCurrentKeyAndKeyGroup(key,
keyGroupIndex) — sets both explicitly, no hashCode recompute (already used by
ChangelogKeyedStateBackend today). KeyedStateInputFormat/MultiStateKeyIterator
already have the discovered key-group in hand when iterating; passing it
through and calling setCurrentKeyAndKeyGroup() instead of setCurrentKey() would
make the read path immune to hashCode drift, with no change to the shared
validation itself.
(Not proposing relaxing the check itself — it's shared with live restore and,
for RocksDB, is the only thing stopping a wrong key-group from being written as
a literal storage-key prefix. The fix above is scoped entirely to the State
Processor API's own key iteration.)
Affected classes: org.apache.flink.state.api.input.KeyedStateInputFormat,
MultiStateKeyIterator (flink-state-processing-api)
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
