[jira] [Updated] (FLINK-40442) 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

["Grzegorz Liter (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-40442?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Grzegorz Liter updated FLINK-40442:
-----------------------------------
    Description: 
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:java}
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)

  was:
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)


> 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
>            Priority: Major
>
> 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:java}
> 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)