You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sr...@apache.org on 2018/02/19 14:08:12 UTC
[19/35] flink git commit: [hotfix][runtime] Simplify RecordWriter code
[hotfix][runtime] Simplify RecordWriter code
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/91dc1c9a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/91dc1c9a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/91dc1c9a
Branch: refs/heads/master
Commit: 91dc1c9ad0bc4430b069ca09b59b750a5166fa86
Parents: e9943c5
Author: Piotr Nowojski <pi...@gmail.com>
Authored: Thu Jan 25 15:19:35 2018 +0100
Committer: Piotr Nowojski <pi...@gmail.com>
Committed: Mon Feb 19 12:21:29 2018 +0100
----------------------------------------------------------------------
.../runtime/io/network/api/writer/RecordWriter.java | 11 +++++------
1 file changed, 5 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/91dc1c9a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
index fa6fbd3..801e6eb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
@@ -130,13 +130,12 @@ public class RecordWriter<T extends IOReadableWritable> {
if (result.isFullRecord()) {
break;
}
- } else {
- BufferBuilder bufferBuilder =
- targetPartition.getBufferProvider().requestBufferBuilderBlocking();
- checkState(!bufferConsumers[targetChannel].isPresent());
- bufferConsumers[targetChannel] = Optional.of(bufferBuilder.createBufferConsumer());
- result = serializer.setNextBufferBuilder(bufferBuilder);
}
+ BufferBuilder bufferBuilder =
+ targetPartition.getBufferProvider().requestBufferBuilderBlocking();
+ checkState(!bufferConsumers[targetChannel].isPresent());
+ bufferConsumers[targetChannel] = Optional.of(bufferBuilder.createBufferConsumer());
+ result = serializer.setNextBufferBuilder(bufferBuilder);
}
checkState(!serializer.hasSerializedData(), "All data should be written at once");
}