[flink] 02/02: [FLINK-33076][network] Reduce serialization overhead of broadcast emit from ChannelSelectorRecordWriter.

[guoweijie]

This is an automated email from the ASF dual-hosted git repository.

guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit e77dc91a59b81c9727fe2d7b3f29dbdd290266a3
Author: Weijie Guo <res...@163.com>
AuthorDate: Fri Sep 8 16:14:00 2023 +0800

    [FLINK-33076][network] Reduce serialization overhead of broadcast emit from 
ChannelSelectorRecordWriter.
---
 .../io/network/api/writer/ChannelSelectorRecordWriter.java     |  2 +-
 .../flink/runtime/io/network/api/writer/RecordWriter.java      | 10 ++++++++++
 2 files changed, 11 insertions(+), 1 deletion(-)

diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/ChannelSelectorRecordWriter.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/ChannelSelectorRecordWriter.java
index 5b756b693b3..e9e19354ca4 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/ChannelSelectorRecordWriter.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/ChannelSelectorRecordWriter.java
@@ -65,7 +65,7 @@ public final class ChannelSelectorRecordWriter<T extends 
IOReadableWritable>
         ByteBuffer serializedRecord = serializeRecord(serializer, record);
         for (int channelIndex = 0; channelIndex < numberOfChannels; 
channelIndex++) {
             serializedRecord.rewind();
-            emit(record, channelIndex);
+            emit(serializedRecord, channelIndex);
         }
 
         if (flushAlways) {
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 57f0a800e37..e9760bdadd3 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
@@ -111,6 +111,16 @@ public abstract class RecordWriter<T extends 
IOReadableWritable> implements Avai
         }
     }
 
+    protected void emit(ByteBuffer record, int targetSubpartition) throws 
IOException {
+        checkErroneous();
+
+        targetPartition.emitRecord(record, targetSubpartition);
+
+        if (flushAlways) {
+            targetPartition.flush(targetSubpartition);
+        }
+    }
+
     public void broadcastEvent(AbstractEvent event) throws IOException {
         broadcastEvent(event, false);
     }