[incubator-celeborn] branch main updated: [CELEBORN-962] Add check DiskInfo#Status in PushDataHandler#checkDiskFull

[zh...@apache.org]

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

zhouky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git


The following commit(s) were added to refs/heads/main by this push:
     new 03fc00e6a [CELEBORN-962] Add check DiskInfo#Status in PushDataHandler#checkDiskFull
03fc00e6a is described below

commit 03fc00e6a6177da3a6c8205e4b47551d4c791ca6
Author: jiaoqingbo <11...@qq.com>
AuthorDate: Thu Sep 14 19:53:28 2023 +0800

    [CELEBORN-962] Add check DiskInfo#Status in PushDataHandler#checkDiskFull
    
    ### What changes were proposed in this pull request?
    
    As Title
    
    ### Why are the changes needed?
    
    As Title
    
    ### Does this PR introduce _any_ user-facing change?
    
    NO
    
    ### How was this patch tested?
    
    PASS GA
    
    Closes #1909 from jiaoqingbo/CELEBORN-962.
    
    Authored-by: jiaoqingbo <11...@qq.com>
    Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
---
 .../apache/celeborn/service/deploy/worker/PushDataHandler.scala  | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
index 36167ad47..68facb601 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
@@ -26,7 +26,7 @@ import io.netty.buffer.ByteBuf
 
 import org.apache.celeborn.common.exception.{AlreadyClosedException, CelebornIOException}
 import org.apache.celeborn.common.internal.Logging
-import org.apache.celeborn.common.meta.{WorkerInfo, WorkerPartitionLocationInfo}
+import org.apache.celeborn.common.meta.{DiskStatus, WorkerInfo, WorkerPartitionLocationInfo}
 import org.apache.celeborn.common.metrics.source.Source
 import org.apache.celeborn.common.network.buffer.{NettyManagedBuffer, NioManagedBuffer}
 import org.apache.celeborn.common.network.client.{RpcResponseCallback, TransportClient, TransportClientFactory}
@@ -1105,9 +1105,12 @@ class PushDataHandler extends BaseMessageHandler with Logging {
     if (fileWriter.flusher.isInstanceOf[HdfsFlusher]) {
       return false
     }
-    val diskFull = workerInfo.diskInfos
+    val diskInfo = workerInfo.diskInfos
       .get(fileWriter.flusher.asInstanceOf[LocalFlusher].mountPoint)
-      .actualUsableSpace < diskReserveSize
+
+    val diskFull = diskInfo.status.equals(
+      DiskStatus.HIGH_DISK_USAGE) || diskInfo.actualUsableSpace < diskReserveSize
+
     diskFull
   }