title: Could not restore Flink keyed state backend from savepoint when upgrading from 1.10 to 1.11
body: <p>We tried to migrate to Flink 1.11 recovering the job from a savepoint taken in 1.10. The job code was not changed, only updated the Flink version of the dependencies to 1.11 (in SBT, we use Scala) and re-built the jar.
  All operators have uids and the job correctly recovers from that savepoint if run on a 1.10 cluster, we are getting the following exception and have no clue:</p>
  <pre><code>java.lang.Exception: Exception while creating StreamOperatorStateContext.
      at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:204)
      at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:247)
      at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:290)
      at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:473)
      at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:47)
      at org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:469)
      at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:522)
      at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721)
      at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546)
      at java.lang.Thread.run(Thread.java:748)
  Caused by: org.apache.flink.util.FlinkException: Could not restore keyed state backend for CoStreamFlatMap_8a6da66867c6cf8469bae55e9f834297_(1/1) from any of the 1 provided restore options.
      at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
      at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:317)
      at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:144)
      ... 9 more
  Caused by: org.apache.flink.runtime.state.BackendBuildingException: Failed when trying to restore heap backend
      at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:116)
      at org.apache.flink.runtime.state.filesystem.FsStateBackend.createKeyedStateBackend(FsStateBackend.java:540)
      at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:301)
      at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:142)
      at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:121)
      ... 11 more
  Caused by: java.lang.IllegalStateException: Missing value for the key 'org.apache.flink.runtime.checkpoint.savepoint.Savepoint'
      at org.apache.flink.util.LinkedOptionalMap.unwrapOptionals(LinkedOptionalMap.java:190)
      at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializerSnapshot.restoreSerializer(KryoSerializerSnapshot.java:86)
      at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
      at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
      at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
      at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
      at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:546)
      at java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260)
      at java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:505)
      at org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate.snapshotsToRestoreSerializers(NestedSerializersSnapshotDelegate.java:225)
      at org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate.getRestoredNestedSerializers(NestedSerializersSnapshotDelegate.java:83)
      at org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot.restoreSerializer(CompositeTypeSerializerSnapshot.java:204)
      at org.apache.flink.runtime.state.StateSerializerProvider.previousSchemaSerializer(StateSerializerProvider.java:189)
      at org.apache.flink.runtime.state.StateSerializerProvider.currentSchemaSerializer(StateSerializerProvider.java:164)
      at org.apache.flink.runtime.state.RegisteredKeyValueStateBackendMetaInfo.getStateSerializer(RegisteredKeyValueStateBackendMetaInfo.java:136)
      at org.apache.flink.runtime.state.heap.StateTable.getStateSerializer(StateTable.java:315)
      at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.createStateMap(CopyOnWriteStateTable.java:54)
      at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.createStateMap(CopyOnWriteStateTable.java:36)
      at org.apache.flink.runtime.state.heap.StateTable.&lt;init&gt;(StateTable.java:98)
      at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.&lt;init&gt;(CopyOnWriteStateTable.java:49)
      at org.apache.flink.runtime.state.heap.AsyncSnapshotStrategySynchronicityBehavior.newStateTable(AsyncSnapshotStrategySynchronicityBehavior.java:41)
      at org.apache.flink.runtime.state.heap.HeapSnapshotStrategy.newStateTable(HeapSnapshotStrategy.java:243)
      at org.apache.flink.runtime.state.heap.HeapRestoreOperation.createOrCheckStateForMetaInfo(HeapRestoreOperation.java:185)
      at org.apache.flink.runtime.state.heap.HeapRestoreOperation.restore(HeapRestoreOperation.java:152)
      at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:114)
  </code></pre>
  <p>Can anyone help?</p>
  <p>Thanks</p>
  <p><strong>UPDATE</strong></p>
  <p>The savepoint comes from a savepoint processed with the stateprocessor API and the state in the <code>KeyedStateBootstrapFunction</code> is made of:</p>
  <pre><code>  var mapToDetector: MapState[String, Map[String, Detector]] = null
    var detectorsConfigs: MapState[String,AnomalyStepConfiguration] = null
    var outputTopic : ValueState[String]= null
    var pipeStatus: MapState[String, String] = null
    var debounceMap: MapState[String, Map[String, DebounceStats]] = null
  </code></pre>