[flink] branch master updated: [FLINK-27522][network] Ignore max buffers per channel when allocate buffer

[pnowojski]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new 91462206889 [FLINK-27522][network] Ignore max buffers per channel when 
allocate buffer
91462206889 is described below

commit 914622068892ee2c5bd60532f3cabec1f581262d
Author: fanrui <1996fan...@gmail.com>
AuthorDate: Sat May 28 15:40:07 2022 +0800

    [FLINK-27522][network] Ignore max buffers per channel when allocate buffer
---
 .../flink/runtime/io/network/buffer/LocalBufferPool.java       |  6 ------
 .../flink/runtime/io/network/buffer/LocalBufferPoolTest.java   | 10 ++++++++--
 2 files changed, 8 insertions(+), 8 deletions(-)

diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
index 4936670e09a..0d5068c02f3 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
@@ -372,12 +372,6 @@ class LocalBufferPool implements BufferPool {
         synchronized (availableMemorySegments) {
             checkDestroyed();
 
-            // target channel over quota; do not return a segment
-            if (targetChannel != UNKNOWN_CHANNEL
-                    && subpartitionBuffersCount[targetChannel] >= 
maxBuffersPerChannel) {
-                return null;
-            }
-
             segment = availableMemorySegments.poll();
 
             if (segment == null) {
diff --git 
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
 
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
index 56e1befddda..7401b5c6e9e 100644
--- 
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
+++ 
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
@@ -453,7 +453,11 @@ public class LocalBufferPoolTest extends TestLogger {
         final BufferBuilder bufferBuilder02 = 
localBufferPool.requestBufferBuilderBlocking(0);
         assertFalse(localBufferPool.getAvailableFuture().isDone());
 
-        assertNull(localBufferPool.requestBufferBuilder(0));
+        // When this channel reaches maxBuffersPerChannel, LocalBufferPool 
will be unavailable. But
+        // when a memory segment is requested from LocalBufferPool and 
availableMemorySegments isn't
+        // empty, maxBuffersPerChannel will be ignored and buffers will 
continue to be allocated.
+        // Check FLINK-27522 for more information.
+        final BufferBuilder bufferBuilder03 = 
localBufferPool.requestBufferBuilderBlocking(0);
         final BufferBuilder bufferBuilder21 = 
localBufferPool.requestBufferBuilderBlocking(2);
         final BufferBuilder bufferBuilder22 = 
localBufferPool.requestBufferBuilderBlocking(2);
         assertFalse(localBufferPool.getAvailableFuture().isDone());
@@ -464,9 +468,11 @@ public class LocalBufferPoolTest extends TestLogger {
         bufferBuilder21.close();
         assertFalse(localBufferPool.getAvailableFuture().isDone());
         bufferBuilder02.close();
-        assertTrue(localBufferPool.getAvailableFuture().isDone());
+        assertFalse(localBufferPool.getAvailableFuture().isDone());
         bufferBuilder01.close();
         assertTrue(localBufferPool.getAvailableFuture().isDone());
+        bufferBuilder03.close();
+        assertTrue(localBufferPool.getAvailableFuture().isDone());
         bufferBuilder22.close();
         assertTrue(localBufferPool.getAvailableFuture().isDone());
     }