[flink] 01/02: [hotfix][network] Flush writing buffers when closing HashSubpartitionBufferAccumulator of tiered storage

[guoweijie]

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

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

commit 59368aedad90cda46883ed0bd0b8fa44aa75c0fe
Author: Yuxin Tan <tanyuxinw...@gmail.com>
AuthorDate: Fri Sep 15 10:31:51 2023 +0800

    [hotfix][network] Flush writing buffers when closing 
HashSubpartitionBufferAccumulator of tiered storage
---
 .../hybrid/tiered/storage/HashSubpartitionBufferAccumulator.java         | 1 +
 1 file changed, 1 insertion(+)

diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/HashSubpartitionBufferAccumulator.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/HashSubpartitionBufferAccumulator.java
index e610926a55d..4a5e9af3ea0 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/HashSubpartitionBufferAccumulator.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/HashSubpartitionBufferAccumulator.java
@@ -75,6 +75,7 @@ public class HashSubpartitionBufferAccumulator {
     }
 
     public void close() {
+        finishCurrentWritingBufferIfNotEmpty();
         while (!unfinishedBuffers.isEmpty()) {
             unfinishedBuffers.poll().close();
         }