[incubator-celeborn] branch main updated: [CELEBORN-928][BUG] Don't stop LocalFlusher when notify error

[zhouky]

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 30496a155 [CELEBORN-928][BUG] Don't stop LocalFlusher when notify error
30496a155 is described below

commit 30496a15503c85cf3d17631de5685739edc46b40
Author: 宪英 <xianying...@antgroup.com>
AuthorDate: Tue Aug 29 22:20:50 2023 +0800

    [CELEBORN-928][BUG] Don't stop LocalFlusher when notify error
    
    ### What changes were proposed in this pull request?
    Don't stop LocalFlusher when notify error.
    
    ### Why are the changes needed?
    
    If LocalDeviceMonitor find non-critical error(e.g. disk full usage) count 
exceeds threshold and notify error to observed device. LocalFlusher will do 
stopAndCleanFlusher, which will interrupt flush threads. Then if the disk 
recover from the error, new disk buffer pushed into the flush queue cannot be 
flushed. It  always causes high memory usage, which would cause 
pausePushDataAndReplicate, and can't recover from it .
    
    ### Does this PR introduce _any_ user-facing change?
    no
    
    ### How was this patch tested?
    Manual.
    
    1. Run a spark job with shuffle.
    2. Copy data in a worker to trigger full disk error.
    3. See logs as below
    
![image](https://github.com/apache/incubator-celeborn/assets/16849112/a4da0bfd-418d-4062-9426-95ef775a9dea)
    5. Confirm flush threads are still running.
    6. Submit another spark job with shuffle, and found disk buffer increase 
and decrease while flushing disk.
    
![image](https://github.com/apache/incubator-celeborn/assets/16849112/7722ebb7-c110-4d39-a87f-d4ab86a25f86)
    
    Closes #1860 from liying919/opt-disk-buffer.
    
    Authored-by: 宪英 <xianying...@antgroup.com>
    Signed-off-by: zky.zhoukeyong <zky.zhoukey...@alibaba-inc.com>
---
 .../org/apache/celeborn/service/deploy/worker/storage/Flusher.scala  | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git 
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/Flusher.scala
 
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/Flusher.scala
index 3b615227f..de3571791 100644
--- 
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/Flusher.scala
+++ 
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/Flusher.scala
@@ -169,9 +169,7 @@ private[worker] class LocalFlusher(
   }
 
   override def notifyError(mountPoint: String, diskStatus: DiskStatus): Unit = 
{
-    logError(s"$this is notified Disk $mountPoint $diskStatus! Stop 
LocalFlusher.")
-    stopAndCleanFlusher()
-    deviceMonitor.unregisterFlusher(this)
+    logError(s"$this is notified Disk $mountPoint $diskStatus! Won't stop 
LocalFlusher.")
   }
 
   override def hashCode(): Int = {
@@ -199,7 +197,6 @@ final private[worker] class HdfsFlusher(
   override def toString: String = s"HdfsFlusher@$flusherId"
 
   override def processIOException(e: IOException, deviceErrorType: 
DiskStatus): Unit = {
-    stopAndCleanFlusher()
     logError(s"$this write failed, reason $deviceErrorType ,exception: $e")
   }