fix read hang

apache · Aug 28, 2023 · d550637 · d550637
1 parent 41b60eb
commit d550637
Showing 1 changed file with 3 additions and 3 deletions.
diff --git a/...c/main/java/org/apache/celeborn/service/deploy/worker/storage/MapDataPartitionReader.java b/...c/main/java/org/apache/celeborn/service/deploy/worker/storage/MapDataPartitionReader.java
@@ -411,9 +411,6 @@ public FileInfo getFileInfo() {
   public void closeReader() {
     synchronized (lock) {
       readFinished = true;
-      // old client can't support BufferStreamEnd, so for new client it tells client that this
-      // stream is finished.
-      if (!isLegacy) associatedChannel.writeAndFlush(new BufferStreamEnd(streamId));
     }
     logger.debug("Closed read for stream {}", this.streamId);
   }
@@ -443,6 +440,9 @@ public void release() {
     synchronized (lock) {
       if (!isReleased) {
         logger.debug("release reader for stream {}", streamId);
+        // old client can't support BufferStreamEnd, so for new client it tells client that this
+        // stream is finished.
+        if (!isLegacy) associatedChannel.writeAndFlush(new BufferStreamEnd(streamId));
         if (!buffersToSend.isEmpty()) {
           numInUseBuffers.addAndGet(-1 * buffersToSend.size());
           buffersToSend.forEach(RecyclableBuffer::recycle);