Aaron He created FLINK-41092:
--------------------------------
Summary: RocksDB restore fails with ClassCastException when
restoring KEY_VALUE_V2 state metadata from async ForSt
Key: FLINK-41092
URL: https://issues.apache.org/jira/browse/FLINK-41092
Project: Flink
Issue Type: Bug
Components: Runtime / State Backends
Reporter: Aaron He
h2. Problem
When a checkpoint produced by the asynchronous ForSt state backend
(ForStKeyedStateBackend, which writes StateMetaInfoSnapshot with
BackendStateType.KEY_VALUE_V2) is restored by RocksDBKeyedStateBackend, state
registration fails with a ClassCastException:
{noformat}
java.lang.ClassCastException: class
org.apache.flink.runtime.state.v2.RegisteredKeyValueStateBackendMetaInfo
cannot be cast to class
org.apache.flink.runtime.state.RegisteredKeyValueStateBackendMetaInfo
at
org.apache.flink.state.rocksdb.RocksDBKeyedStateBackend.tryRegisterKvStateInformation(RocksDBKeyedStateBackend.java:798)
{noformat}
h2. Cause
ForStIncrementalRestoreOperation already converts restored state metadata
snapshots via
RegisteredKeyValueStateBackendMetaInfo.fromMetaInfoSnapshot(stateMetaInfoSnapshot),
which accepts both KEY_VALUE and KEY_VALUE_V2 snapshots and constructs a V1
RegisteredKeyValueStateBackendMetaInfo (converting V2 MAP state's
UserKeySerializer and value StateSerializer into a V1 MapSerializer).
By contrast, RocksDB's three restore paths
(RocksDBHandle#getOrRegisterStateColumnFamilyHandle,
RestoredDBInstance#createColumnFamilyDescriptors, and
DistributeStateHandlerHelper) call
RegisteredStateMetaInfoBase.fromMetaInfoSnapshot(stateMetaInfoSnapshot), which
returns
org.apache.flink.runtime.state.v2.RegisteredKeyValueStateBackendMetaInfo for
KEY_VALUE_V2 snapshots. RocksDBKeyedStateBackend expects V1
RegisteredKeyValueStateBackendMetaInfo instances in kvStateInformation.
h2. Proposal
Use
RegisteredKeyValueStateBackendMetaInfo.fromMetaInfoSnapshot(stateMetaInfoSnapshot)
in RocksDBHandle, RestoredDBInstance, and DistributeStateHandlerHelper,
matching ForStIncrementalRestoreOperation. Priority-queue snapshots are already
separated in RocksDBIncrementalRestoreOperation#readMetaData before reaching
these methods.
h2. Public source
[RocksDBHandle.java|https://github.com/apache/flink/blob/855597a5460630eb8f72510277bfed7c0be0a004/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/restore/RocksDBHandle.java#L176-L178]
[RestoredDBInstance.java|https://github.com/apache/flink/blob/855597a5460630eb8f72510277bfed7c0be0a004/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/restore/RestoredDBInstance.java#L151-L153]
[DistributeStateHandlerHelper.java|https://github.com/apache/flink/blob/855597a5460630eb8f72510277bfed7c0be0a004/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/restore/DistributeStateHandlerHelper.java#L137-L140]
[RocksDBKeyedStateBackend.java|https://github.com/apache/flink/blob/855597a5460630eb8f72510277bfed7c0be0a004/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackend.java#L796-L799]
h2. Tests and acceptance
Verify that KEY_VALUE_V2 state metadata snapshots (Value, List, Map, Reducing,
Aggregating) restore into RocksDBKeyedStateBackend across same-parallelism,
scale-up, and scale-down (copy and ingest-DB restore modes), with restored
values readable and unaccessed states retained in subsequent snapshots.
{noformat}
./mvnw -pl flink-state-backends/flink-statebackend-rocksdb
-Dtest=RocksIncrementalCheckpointRescalingTest,DistributeStateHandlerHelperTest
test
{noformat}
h2. Proposed release note
RocksDBKeyedStateBackend can now restore keyed state metadata snapshots written
by the asynchronous ForSt state backend (KEY_VALUE_V2).
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
