[hotfix][network] Rename ResultPartitionWriter#initializeState to #re…

…adRecoveredState

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/ResultPartitionWriter.java

@@ -45,10 +45,10 @@ public interface ResultPartitionWriter extends AutoCloseable, AvailabilityProvid
 	void setup() throws IOException;
 
 	/**
-	 * Loads the previous output states with the given reader for unaligned checkpoint.
+	 * Reads the previous output states with the given reader for unaligned checkpoint.
 	 * It should be done before task processing the inputs.
 	 */
-	void initializeState(ChannelStateReader stateReader) throws IOException, InterruptedException;
+	void readRecoveredState(ChannelStateReader stateReader) throws IOException, InterruptedException;
 
 	ResultPartitionID getPartitionId();
 

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java

@@ -102,7 +102,7 @@ public class PipelinedSubpartition extends ResultSubpartition {
 	}
 
 	@Override
-	public void initializeState(ChannelStateReader stateReader) throws IOException, InterruptedException {
+	public void readRecoveredState(ChannelStateReader stateReader) throws IOException, InterruptedException {
 		for (ReadResult readResult = ReadResult.HAS_MORE_DATA; readResult == ReadResult.HAS_MORE_DATA;) {
 			BufferBuilder bufferBuilder = parent.getBufferPool().requestBufferBuilderBlocking(subpartitionInfo.getSubPartitionIdx());
 			BufferConsumer bufferConsumer = bufferBuilder.createBufferConsumer();

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java

@@ -152,9 +152,9 @@ public void setup() throws IOException {
 	}
 
 	@Override
-	public void initializeState(ChannelStateReader stateReader) throws IOException, InterruptedException {
+	public void readRecoveredState(ChannelStateReader stateReader) throws IOException, InterruptedException {
 		for (ResultSubpartition subpartition : subpartitions) {
-			subpartition.initializeState(stateReader);
+			subpartition.readRecoveredState(stateReader);
 		}
 	}
 

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartition.java

@@ -77,7 +77,7 @@ protected void onConsumedSubpartition() {
 		parent.onConsumedSubpartition(getSubPartitionIndex());
 	}
 
-	public void initializeState(ChannelStateReader stateReader) throws IOException, InterruptedException {
+	public void readRecoveredState(ChannelStateReader stateReader) throws IOException, InterruptedException {
 	}
 
 	/**

flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/ConsumableNotifyingResultPartitionWriterDecorator.java

@@ -101,8 +101,8 @@ public ResultSubpartition getSubpartition(int subpartitionIndex) {
 	}
 
 	@Override
-	public void initializeState(ChannelStateReader stateReader) throws IOException, InterruptedException {
-		partitionWriter.initializeState(stateReader);
+	public void readRecoveredState(ChannelStateReader stateReader) throws IOException, InterruptedException {
+		partitionWriter.readRecoveredState(stateReader);
 	}
 
 	@Override

flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterTest.java

@@ -463,7 +463,7 @@ public void testEmitRecordWithPartitionStateRecovery() throws Exception {
 
 		try {
 			partition.setup();
-			partition.initializeState(stateReader);
+			partition.readRecoveredState(stateReader);
 
 			for (int record: records) {
 				// the record length 4 is also written into buffer for every emit

flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/MockResultPartitionWriter.java

@@ -40,7 +40,7 @@ public void setup() {
 	}
 
 	@Override
-	public void initializeState(ChannelStateReader stateReader) {
+	public void readRecoveredState(ChannelStateReader stateReader) {
 	}
 
 	@Override

flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java

@@ -453,7 +453,7 @@ public void testInitializeEmptyState() throws Exception {
 		final ChannelStateReader stateReader = ChannelStateReader.NO_OP;
 		try {
 			partition.setup();
-			partition.initializeState(stateReader);
+			partition.readRecoveredState(stateReader);
 
 			for (ResultSubpartition subpartition : partition.getAllPartitions()) {
 				// no buffers are added into the queue for empty states
@@ -513,7 +513,7 @@ public void testInitializeMoreStateThanBuffer() throws Exception {
 			Future<Void> result = executor.submit(partitionConsumeTask);
 
 			partition.setup();
-			partition.initializeState(stateReader);
+			partition.readRecoveredState(stateReader);
 
 			// wait the partition consume task finish
 			result.get(20, TimeUnit.SECONDS);

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java

@@ -463,7 +463,7 @@ protected void beforeInvoke() throws Exception {
 			if (writers != null) {
 				//TODO we should get proper state reader from getEnvironment().getTaskStateManager().getChannelStateReader()
 				for (ResultPartitionWriter writer : writers) {
-					writer.initializeState(ChannelStateReader.NO_OP);
+					writer.readRecoveredState(ChannelStateReader.NO_OP);
 				}
 			}
 		});

flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java

@@ -1776,7 +1776,7 @@ private static class RecoveryResultPartition extends MockResultPartitionWriter {
 		}
 
 		@Override
-		public void initializeState(ChannelStateReader stateReader) {
+		public void readRecoveredState(ChannelStateReader stateReader) {
 			isStateInitialized = true;
 		}