[
https://issues.apache.org/jira/browse/FLINK-9169?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16462554#comment-16462554
]
ASF GitHub Bot commented on FLINK-9169:
---------------------------------------
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/5950#discussion_r185828137
--- Diff:
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSerializationUtil.java
---
@@ -200,7 +197,7 @@ public static void
writeSerializersAndConfigsWithResilience(
for (int i = 0; i < numSerializersAndConfigSnapshots;
i++) {
bufferWithPos.setPosition(offsets[i * 2]);
- serializer = tryReadSerializer(bufferWrapper,
userCodeClassLoader);
+ serializer = tryReadSerializer(bufferWrapper,
userCodeClassLoader, true);
--- End diff --
The problem is that this method might mix too many things together, that is
also again visible in the complex return type and e.g. many call sites are only
interested in the first element of the list. Wonder if we should break this up
in dedicated steps (serializer, config) and let the callers invoke them one by
one, so that we can handle exceptions on a higher level and make decisions
about if we need to have a serializer there.
> NPE when restoring from old savepoint and state serializer could not be
> deserialized
> ------------------------------------------------------------------------------------
>
> Key: FLINK-9169
> URL: https://issues.apache.org/jira/browse/FLINK-9169
> Project: Flink
> Issue Type: Bug
> Components: State Backends, Checkpointing
> Affects Versions: 1.5.0, 1.4.2
> Reporter: Till Rohrmann
> Assignee: Tzu-Li (Gordon) Tai
> Priority: Major
> Fix For: 1.5.0
>
>
> A user reported to have observed the following exception when restoring a
> Flink job from a 1.3 savepoint with Flink 1.4.
> {code}
> 2018-04-02 21:44:18,146 INFO org.apache.flink.runtime.taskmanager.Task
> - ApplyAMUpdate (13/160) (7248adb0b85f4458ae4144963d65
> 6fa6) switched from RUNNING to FAILED.
> java.lang.IllegalStateException: Could not initialize keyed state backend.
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.initKeyedState(AbstractStreamOperator.java:293)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:225)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.initializeOperators(StreamTask.java:692)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.initializeState(StreamTask.java:679)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:253)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:718)
> at java.lang.Thread.run(Thread.java:748)
> Caused by: java.lang.NullPointerException
> at
> org.apache.flink.util.Preconditions.checkNotNull(Preconditions.java:58)
> at
> org.apache.flink.runtime.state.RegisteredKeyedBackendStateMetaInfo.<init>(RegisteredKeyedBackendStateMetaInfo.java:53)
> at
> org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend$RocksDBFullRestoreOperation.restoreKVStateMetaData(RocksDBKeyedStateB
> ackend.java:1216)
> at
> org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend$RocksDBFullRestoreOperation.restoreKeyGroupsInStateHandle(RocksDBKeye
> dStateBackend.java:1153)
> at
> org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend$RocksDBFullRestoreOperation.doRestore(RocksDBKeyedStateBackend.java:1
> 139)
> at
> org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend.restore(RocksDBKeyedStateBackend.java:1034)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.createKeyedStateBackend(StreamTask.java:773)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.initKeyedState(AbstractStreamOperator.java:283)
> ... 6 more
> {code}
> Looking at the {{KeyedBackendStateMetaInfoSnapshotReaderWriters}}, we create
> {{RegisteredKeyedBackendStateMetaInfo.Snapshot}} where the
> {{stateSerializer}} can be {{null}}. This is not the problem, however, in
> {{RocksDBKeyedStateBackend#restoreKVStateMetaData}} we create a
> {{RegisteredKeyedBackendStateMetaInfo}} from the deserialized {{Snapshot}}
> where we null check the state serializer. This will then fail with an
> indescriptive NPE.
> I think the same should happen when resuming with Flink 1.5 from a 1.4
> savepoint.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
