Github user yanghua commented on a diff in the pull request: https://github.com/apache/flink/pull/5723#discussion_r175729909 --- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java --- @@ -279,15 +279,14 @@ protected OperatorStateBackend operatorStateBackend( if (restoreStateAlternatives.hasNext()) { - final CloseableRegistry closeableRegistry = new CloseableRegistry(); - Collection<OperatorStateHandle> rawOperatorState = restoreStateAlternatives.next(); // TODO currently this does not support local state recovery, so we expect there is only one handle. Preconditions.checkState( !restoreStateAlternatives.hasNext(), "Local recovery is currently not implemented for raw operator state, but found state alternative."); if (rawOperatorState != null) { + final CloseableRegistry closeableRegistry = new CloseableRegistry(); --- End diff -- Yes. Obviously, we could do this. Let me refactor it.
---