(flink) branch master updated: [FLINK-40525][network] Remove residual receivedBuffers lock in finishReadRecoveredState to avoid ABBA deadlock

[fanrui]

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

1996fanrui 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 c7489bc5eb8 [FLINK-40525][network] Remove residual receivedBuffers 
lock in finishReadRecoveredState to avoid ABBA deadlock
c7489bc5eb8 is described below

commit c7489bc5eb892659482b5ce88f7dbcb532693598
Author: Rui Fan <1996fan...@gmail.com>
AuthorDate: Mon Aug 31 18:46:37 2026 +0200

    [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 behavio [...]
---
 .../io/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 4714c4ef28d..3ff342a3178 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 abstract class RecoveredInputChannel extends 
InputChannel implements Chan
     }
 
     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);
     }