From a6e89f22ce85196fa931343ab589390e5cf407fa Mon Sep 17 00:00:00 2001
From: Rui Fan <1996fanrui@gmail.com>
Date: Mon, 31 Aug 2026 18:46:37 +0200
Subject: [PATCH] [FLINK-40525][network] Remove residual receivedBuffers lock
 in finishReadRecoveredState to avoid ABBA deadlock

finishReadRecoveredState() wrapped onRecoveredStateBuffer() in synchronized(receivedBuffers), but that method already locks receivedBuffers internally and, when the queue was empty, calls notifyChannelNonEmpty(), which acquires inputChannelsWithData while receivedBuffers is still held. This inverts the task thread's inputChannelsWithData -> receivedBuffers lock order and can deadlock on the error path (ABBA). Removing the redundant outer lock restores the correct order with no behavior change, since onRecoveredStateBuffer already provides the required mutual exclusion.
---
 .../network/partition/consumer/RecoveredInputChannel.java   | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java
index 4714c4ef28d7af..3ff342a3178233 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java
@@ -161,10 +161,8 @@ public void onRecoveredStateBuffer(Buffer buffer) {
     }
 
     public void finishReadRecoveredState() throws IOException {
-        synchronized (receivedBuffers) {
-            onRecoveredStateBuffer(
-                    EventSerializer.toBuffer(EndOfInputChannelStateEvent.INSTANCE, false));
-        }
+        onRecoveredStateBuffer(
+                EventSerializer.toBuffer(EndOfInputChannelStateEvent.INSTANCE, false));
         bufferManager.releaseFloatingBuffers();
         LOG.debug("{}/{} finished recovering input.", inputGate.getOwningTaskName(), channelInfo);
     }