[
https://issues.apache.org/jira/browse/FLINK-11287?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Tzu-Li (Gordon) Tai updated FLINK-11287:
----------------------------------------
Description:
Currently, when creating a {{RocksDBListState}}, the element serializer wrapped
by the {{RocksDBListState}} is retrieved from the state descriptor:
{code:java}
static <E, K, N, SV, S extends State, IS extends S> IS create(
StateDescriptor<S, SV> stateDesc,
Tuple2<ColumnFamilyHandle, RegisteredKeyValueStateBackendMetaInfo<N, SV>>
registerResult,
RocksDBKeyedStateBackend<K> backend) {
return (IS) new RocksDBListState<>(
registerResult.f0,
registerResult.f1.getNamespaceSerializer(),
(TypeSerializer<List<E>>) registerResult.f1.getStateSerializer(),
(List<E>) stateDesc.getDefaultValue(),
backend);
}
{code}
This is incorrect, since new serializers retrieved from state descriptors have
not been checked for compatibility with restored state.
Instead, the element serializer should be retrieved from the register result,
which contains the actual state serializer registered in the state backend for
state access (and has already been checked for compatibility / reconfigured
appropriately).
was:
Currently, when creating a {{RocksDBListState}}, the element serializer wrapped
by the {{RocksDBListState}} is retrieved from the state descriptor:
{code}
static <E, K, N, SV, S extends State, IS extends S> IS create(
StateDescriptor<S, SV> stateDesc,
Tuple2<ColumnFamilyHandle, RegisteredKeyValueStateBackendMetaInfo<N, SV>>
registerResult,
RocksDBKeyedStateBackend<K> backend) {
return (IS) new RocksDBListState<>(
registerResult.f0,
registerResult.f1.getNamespaceSerializer(),
(TypeSerializer<List<E>>) registerResult.f1.getStateSerializer(),
(List<E>) stateDesc.getDefaultValue(),
backend);
}
{code}
This is incorrect, since new serializers retrieved from state descriptors have
not been checked for compatibility with restored state.
Instead, the element serializer should be retrieved from the register result,
which contains the actual state serializer registered in the state backend for
state access (and has already been checked for compatibility / reconfigured
appropriately).
> RocksDBListState creation should use actual registered state serializer
> instead of serializer provided by descriptor
> --------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-11287
> URL: https://issues.apache.org/jira/browse/FLINK-11287
> Project: Flink
> Issue Type: Bug
> Components: State Backends, Checkpointing
> Reporter: Tzu-Li (Gordon) Tai
> Assignee: Tzu-Li (Gordon) Tai
> Priority: Critical
> Fix For: 1.8.0
>
>
> Currently, when creating a {{RocksDBListState}}, the element serializer
> wrapped by the {{RocksDBListState}} is retrieved from the state descriptor:
> {code:java}
> static <E, K, N, SV, S extends State, IS extends S> IS create(
> StateDescriptor<S, SV> stateDesc,
> Tuple2<ColumnFamilyHandle, RegisteredKeyValueStateBackendMetaInfo<N, SV>>
> registerResult,
> RocksDBKeyedStateBackend<K> backend) {
> return (IS) new RocksDBListState<>(
> registerResult.f0,
> registerResult.f1.getNamespaceSerializer(),
> (TypeSerializer<List<E>>) registerResult.f1.getStateSerializer(),
> (List<E>) stateDesc.getDefaultValue(),
> backend);
> }
> {code}
> This is incorrect, since new serializers retrieved from state descriptors
> have not been checked for compatibility with restored state.
> Instead, the element serializer should be retrieved from the register result,
> which contains the actual state serializer registered in the state backend
> for state access (and has already been checked for compatibility /
> reconfigured appropriately).
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
