[incubator-celeborn] branch branch-0.3 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 branch-0.3
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git


The following commit(s) were added to refs/heads/branch-0.3 by this push:
     new 7de2c9ab5 [CELEBORN-928][BUG] Don't stop LocalFlusher when notify error
7de2c9ab5 is described below

commit 7de2c9ab572d099350fcfe227af4f82a9edc490a
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>
    (cherry picked from commit 30496a15503c85cf3d17631de5685739edc46b40)
    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")
   }