[hotfix][checkpointing] Use practical ChannelStateReader instead of N…

…O_OP
is · Apr 29, 2020 · 59ae237 · 59ae237
1 parent 4a1363f
commit 59ae237
Showing 1 changed file with 1 addition and 3 deletions.
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -28,7 +28,6 @@
 import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
 import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
-import org.apache.flink.runtime.checkpoint.channel.ChannelStateReader;
 import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
 import org.apache.flink.runtime.concurrent.FutureUtils;
 import org.apache.flink.runtime.execution.CancelTaskException;
@@ -461,9 +460,8 @@ protected void beforeInvoke() throws Exception {
 
  ResultPartitionWriter[] writers = getEnvironment().getAllWriters();
  if (writers != null) {
- //TODO we should get proper state reader from getEnvironment().getTaskStateManager().getChannelStateReader()
  for (ResultPartitionWriter writer : writers) {
- writer.readRecoveredState(ChannelStateReader.NO_OP);
+ writer.readRecoveredState(getEnvironment().getTaskStateManager().getChannelStateReader());
  }
  }
  });