From bc80c80fd62c75849ae019f1fbcfc70547fb12dd Mon Sep 17 00:00:00 2001
From: Rui Fan <1996fanrui@gmail.com>
Date: Mon, 31 Aug 2026 18:57:52 +0200
Subject: [PATCH] [FLINK-40524][network] Skip the recoveredBuffers monitor on
 the getNextBuffer hot path for channels that never recover

---
 .../partition/consumer/LocalInputChannel.java | 45 ++++++++++---------
 1 file changed, 23 insertions(+), 22 deletions(-)

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
index 9a7aa4b96dfc15..2e6a173c3d33d9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
@@ -559,31 +559,32 @@ protected int peekNextBufferSubpartitionIdInternal() throws IOException {
     public Optional<BufferAndAvailability> getNextBuffer() throws IOException {
         checkError();
 
-        // Read inRecovery and poll the recovered buffer under a single lock acquisition to avoid
-        // grabbing the monitor twice on the hot path.
-        boolean inRecovery;
-        Buffer recoveredBuf = null;
-        synchronized (recoveredBuffers) {
-            inRecovery = this.inRecovery;
-            if (inRecovery && !hasPendingPriorityEvent && !recoveredBuffers.isEmpty()) {
-                recoveredBuf = recoveredBuffers.poll();
+        if (needsRecovery) {
+            // Read inRecovery and poll the recovered buffer under a single lock acquisition to
+            // avoid grabbing the monitor twice on the hot path.
+            boolean inRecovery;
+            Buffer recoveredBuf = null;
+            synchronized (recoveredBuffers) {
+                inRecovery = this.inRecovery;
+                if (inRecovery && !hasPendingPriorityEvent && !recoveredBuffers.isEmpty()) {
+                    recoveredBuf = recoveredBuffers.poll();
+                }
             }
-        }
 
-        if (inRecovery) {
-            // Always return an already-polled recovered buffer first: hasPendingPriorityEvent may
-            // be flipped to true by a concurrent notifyPriorityEvent() after the poll, and
-            // re-reading
-            // it here would otherwise drop this buffer. A pending priority event is served on the
-            // next getNextBuffer() call instead.
-            if (recoveredBuf != null) {
-                return wrapRecoveredBufferAsAvailability(recoveredBuf);
-            }
-            if (hasPendingPriorityEvent) {
-                return pullPriorityFromSubpartitionView();
+            if (inRecovery) {
+                // Always return an already-polled recovered buffer first: hasPendingPriorityEvent
+                // may be flipped to true by a concurrent notifyPriorityEvent() after the poll, and
+                // re-reading it here would otherwise drop this buffer. A pending priority event is
+                // served on the next getNextBuffer() call instead.
+                if (recoveredBuf != null) {
+                    return wrapRecoveredBufferAsAvailability(recoveredBuf);
+                }
+                if (hasPendingPriorityEvent) {
+                    return pullPriorityFromSubpartitionView();
+                }
+                // Drain not finished yet; block normal upstream data until delivery completes.
+                return Optional.empty();
             }
-            // Drain not finished yet; block normal upstream data until delivery completes.
-            return Optional.empty();
         }
 
         if (!toBeConsumedBuffers.isEmpty()) {