curcur commented on a change in pull request #13614:
URL: https://github.com/apache/flink/pull/13614#discussion_r507171053
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BufferWritingResultPartition.java
##########
@@ -211,46 +223,83 @@ protected void releaseInternal() {
}
}
- private BufferBuilder getSubpartitionBufferBuilder(int
targetSubpartition) throws IOException {
- final BufferBuilder bufferBuilder =
subpartitionBufferBuilders[targetSubpartition];
- if (bufferBuilder != null) {
- return bufferBuilder;
+ private BufferBuilder appendDataToSubpartitionForNewRecord(
+ ByteBuffer record, int targetSubpartition) throws
IOException {
+ BufferBuilder buffer =
subpartitionBufferBuilders[targetSubpartition];
+
+ if (buffer == null) {
+ buffer =
requestNewSubpartitionBufferBuilder(targetSubpartition);
+
subpartitions[targetSubpartition].add(buffer.createBufferConsumerFromBeginning(),
0);
Review comment:
I've actually tried a bit to remove `createBufferConsumerFromBeginning`,
which can work, but the clean-up logic is not as simple and beautful as it is
now: `skip() + build()`. (Reason: skip can be unsuccessful because a partial
record may not been written or visible yet before skip(), but can be visible
after skip()).
So I will keep the way we are going right now.
To your question:
`could it be createBufferConsumer()?`
The answer is yes to create a new record, but I think the meaning is more
clear to use `createBufferConsumerFromBeginning`. What do you think?
I can definitely change it to `createBufferConsumer()`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
