Tzu-Li (Gordon) Tai created FLINK-11094:
-------------------------------------------
Summary: Restored state in RocksDBStateBackend that has not been
accessed in new execution causes NPE on snapshot
Key: FLINK-11094
URL: https://issues.apache.org/jira/browse/FLINK-11094
Project: Flink
Issue Type: Bug
Components: State Backends, Checkpointing
Affects Versions: 1.7.0
Reporter: Tzu-Li (Gordon) Tai
Assignee: Tzu-Li (Gordon) Tai
Fix For: 1.7.1
This was caused by changes in FLINK-10679.
The problem is that in that change, in the {{RocksDBKeyedBackend}},
{{RegisteredStateMetaInfoBase}}s were no longer created eagerly for all
restored state, but instead only lazily created when the state was accessed
again by the user. This causes non-accessed restored state to have empty meta
info, and throws NPE when trying to take a snapshot of them.
The rationale behind FLINK-10679 was that, since
{{RegisteredStateMetaInfoBase}} holds already serializer instances for state
access, creating them eagerly at restore time with restored serializer
snapshots did not make sense (because at that point-in-time, we do not have the
new serializers yet for state access; the snapshot is only capable of creating
the previous state serializer).
I propose the following:
Instead of having final {{TypeSerializer}} instances in
{{RegisteredStateMetaInfoBase}}s, they should have a
{{StateSerializerProvider}} instead.
The {{StateSerializerProvider}} would have the following methods:
{code}
public class StateSerializerProvider<T> {
TypeSerializer<T> getCurrentSerializer();
TypeSerializer<T> updateCurrentSerializer(TypeSerializer<T> newSerializer);
TypeSerializer<T> getPreviousSerializer();
}
{code}
A {{StateSerializerProvider}} can be created either from:
1) A restored serializer snapshot when restoring the state.
2) A fresh, new state's serializer, when registering the state for the first
time.
For 1), state that has not been accessed after the restore will return the same
serializer (i.e. the previous serializer) for both {{getPreviousSerializer}}
and {{getCurrentSerializer}}. Once a restored state is re-accessed, then
{{updateCurrentSerializer(TypeSerializer<T> newSerializer)}} should be used to
update what serializer the provider returns in {{getCurrentSerializer}}.
We could also make use of this new abstraction to move away some of the new
serializer's compatibility checks from the state backend to
{{StateSerializerProvider#updateCurrentSerializer}}.
For tests, apparently we're lacking test coverage for restored state that has
not been accessed and being snapshotted again. This should be included as part
of the fix.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
