AHeise commented on a change in pull request #7713: [FLINK-10995][runtime] Copy
intermediate serialization results only once for broadcast mode
URL: https://github.com/apache/flink/pull/7713#discussion_r327062898
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
##########
@@ -225,46 +178,56 @@ public void setMetricGroup(TaskIOMetricGroup metrics) {
numBuffersOut = metrics.getNumBuffersOutCounter();
}
- /**
- * Marks the current {@link BufferBuilder} as finished and clears the
state for next one.
- */
- private void tryFinishCurrentBufferBuilder(int targetChannel) {
- if (!bufferBuilders[targetChannel].isPresent()) {
- return;
- }
- BufferBuilder bufferBuilder =
bufferBuilders[targetChannel].get();
- bufferBuilders[targetChannel] = Optional.empty();
+ protected void finishBufferBuilder(BufferBuilder bufferBuilder) {
numBytesOut.inc(bufferBuilder.finish());
numBuffersOut.inc();
}
+ /**
+ * This is used to send regular records.
+ */
+ public abstract void emit(T record) throws IOException,
InterruptedException;
+
+ /**
+ * This is used to send LatencyMarks to a random target channel.
+ */
+ public abstract void randomEmit(T record) throws IOException,
InterruptedException;
+
+ /**
+ * This is used to broadcast streaming Watermarks in-band with records.
+ */
+ public abstract void broadcastEmit(T record) throws IOException,
InterruptedException;
+
/**
* The {@link BufferBuilder} may already exist if not filled up last
time, otherwise we need
* request a new one for this target channel.
*/
- private BufferBuilder getBufferBuilder(int targetChannel) throws
IOException, InterruptedException {
- if (bufferBuilders[targetChannel].isPresent()) {
- return bufferBuilders[targetChannel].get();
- } else {
- return requestNewBufferBuilder(targetChannel);
- }
- }
+ abstract BufferBuilder getBufferBuilder(int targetChannel) throws
IOException, InterruptedException;
- private BufferBuilder requestNewBufferBuilder(int targetChannel) throws
IOException, InterruptedException {
- checkState(!bufferBuilders[targetChannel].isPresent() ||
bufferBuilders[targetChannel].get().isFinished());
+ /**
+ * Requests a new {@link BufferBuilder} for the target channel and
returns it.
+ */
+ abstract BufferBuilder requestNewBufferBuilder(int targetChannel)
throws IOException, InterruptedException;
- BufferBuilder bufferBuilder =
targetPartition.getBufferBuilder();
- bufferBuilders[targetChannel] = Optional.of(bufferBuilder);
-
targetPartition.addBufferConsumer(bufferBuilder.createBufferConsumer(),
targetChannel);
- return bufferBuilder;
- }
+ /**
+ * Marks the current {@link BufferBuilder} as finished if present and
clears the state for next one.
+ */
+ abstract void tryFinishCurrentBufferBuilder(int targetChannel);
- private void closeBufferBuilder(int targetChannel) {
- if (bufferBuilders[targetChannel].isPresent()) {
- bufferBuilders[targetChannel].get().finish();
- bufferBuilders[targetChannel] = Optional.empty();
- }
- }
+ /**
+ * Marks the current {@link BufferBuilder} as empty for the target
channel.
+ */
+ abstract void emptyCurrentBufferBuilder(int targetChannel);
+
+ /**
+ * Marks the current {@link BufferBuilder} as finished and empty for
the target channel.
Review comment:
suggestion: Marks the current {@link BufferBuilder} as finished and releases
the resources for the target channel.
----------------------------------------------------------------
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]
With regards,
Apache Git Services
