Currently in the AbstractStreamOperator::initializeState(OperatorStateHandles stateHandles), the restoreStreamCheckpointed which is responsible for restoring state from previous Flink versions, (backwards compatibility) is called before the initializeState(StateInitializationContext context) which is responsible for initializing the state in Flink 1.2.
This has the negative side effect that when implementing the backwards compatibility strategy for a given operator, we have to restore the old state, store it in local variables, and register it with the new state abstractions in the initializeState() or the open(). This creates a lot of unnecessary code in the operators, and potential memory leaks if the local variables are not "null-ified".
This issue proposes to call the restoreStreamCheckpointed after the initializeState(StateInitializationContext context). This way, the new operator state will have been initialized (e.g. keyed state), and the restoreStreamCheckpointed will be able to register its state directly with the new abstractions, instead of putting it in local variables and wait for the initializeState or the open() to re-register it.