Github user pnowojski commented on a diff in the pull request:

    https://github.com/apache/flink/pull/6272#discussion_r201750461
  
    --- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
 ---
    @@ -360,32 +360,44 @@ public boolean notifyBufferAvailable(Buffer buffer) {
                        return false;
                }
     
    -           boolean needMoreBuffers = false;
    -           synchronized (bufferQueue) {
    -                   checkState(isWaitingForFloatingBuffers, "This channel 
should be waiting for floating buffers.");
    +           boolean recycleBuffer = true;
    +           try {
    +                   boolean needMoreBuffers = false;
    +                   synchronized (bufferQueue) {
    +                           checkState(isWaitingForFloatingBuffers,
    +                                   "This channel should be waiting for 
floating buffers.");
    +
    +                           // Important: double check the isReleased state 
inside synchronized block, so there is no
    +                           // race condition when notifyBufferAvailable 
and releaseAllResources running in parallel.
    +                           if (isReleased.get() ||
    +                                   bufferQueue.getAvailableBufferSize() >= 
numRequiredBuffers) {
    --- End diff --
    
    nitty nit: this folding doesn't look good and melds with if's body.


---

Reply via email to