zhijiangW commented on a change in pull request #7713: [FLINK-10995][network]
Copy intermediate serialization results only once for broadcast mode
URL: https://github.com/apache/flink/pull/7713#discussion_r265620648
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
##########
@@ -40,26 +40,33 @@
private final SettablePositionMarker positionMarker = new
SettablePositionMarker();
- private boolean bufferConsumerCreated = false;
-
public BufferBuilder(MemorySegment memorySegment, BufferRecycler
recycler) {
this.memorySegment = checkNotNull(memorySegment);
this.recycler = checkNotNull(recycler);
}
/**
- * @return created matching instance of {@link BufferConsumer} to this
{@link BufferBuilder}. There can exist only
- * one {@link BufferConsumer} per each {@link BufferBuilder} and vice
versa.
+ * @return created matching instance of {@link BufferConsumer} to this
{@link BufferBuilder}.
*/
public BufferConsumer createBufferConsumer() {
- checkState(!bufferConsumerCreated, "There can not exists two
BufferConsumer for one BufferBuilder");
- bufferConsumerCreated = true;
return new BufferConsumer(
memorySegment,
recycler,
positionMarker);
}
+ /**
+ * @return created matching instance of {@link BufferConsumer} similar
with {@link #createBufferConsumer()},
+ * except for initializing its reader position based on {@link
BufferBuilder}'s current writable position .
+ */
+ public BufferConsumer createPositionedBufferConsumer() {
Review comment:
Yes, I would supplement a unit test for it.
----------------------------------------------------------------
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:
us...@infra.apache.org
With regards,
Apache Git Services
