(incubator-celeborn) branch branch-0.4 updated: [CELEBORN-1248] Improve flusher's robustness

[ethanfeng]

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

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


The following commit(s) were added to refs/heads/branch-0.4 by this push:
     new bdf277b52 [CELEBORN-1248] Improve flusher's robustness
bdf277b52 is described below

commit bdf277b5258b9b3a2b305f213ab3b87d4427c278
Author: mingji <fengmingxiao....@alibaba-inc.com>
AuthorDate: Mon Jan 29 21:22:17 2024 +0800

    [CELEBORN-1248] Improve flusher's robustness
    
    ### What changes were proposed in this pull request?
    Improve flusher's robustness.
    
    ### Why are the changes needed?
    Flusher's thread might be terminated due to uncaught exceptions.
    
    ### Does this PR introduce _any_ user-facing change?
    NO.
    
    ### How was this patch tested?
    GA and cluaster.
    
    Closes #2254 from FMX/b1248.
    
    Authored-by: mingji <fengmingxiao....@alibaba-inc.com>
    Signed-off-by: mingji <fengmingxiao....@alibaba-inc.com>
    (cherry picked from commit 99da6e2028a980936d8dfea4dedf56726a37151c)
    Signed-off-by: mingji <fengmingxiao....@alibaba-inc.com>
---
 .../celeborn/service/deploy/worker/storage/Flusher.scala | 16 ++++++++++------
 1 file changed, 10 insertions(+), 6 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 c97b6defc..ffee36f0c 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
@@ -31,7 +31,7 @@ import org.apache.celeborn.common.internal.Logging
 import org.apache.celeborn.common.meta.{DiskStatus, TimeWindow}
 import org.apache.celeborn.common.metrics.source.{AbstractSource, 
ThreadPoolSource}
 import org.apache.celeborn.common.protocol.StorageInfo
-import org.apache.celeborn.common.util.ThreadUtils
+import org.apache.celeborn.common.util.{ThreadUtils, Utils}
 import org.apache.celeborn.service.deploy.worker.WorkerSource
 import 
org.apache.celeborn.service.deploy.worker.congestcontrol.CongestionController
 import org.apache.celeborn.service.deploy.worker.memory.MemoryManager
@@ -76,14 +76,18 @@ abstract private[worker] class Flusher(
                     flushTimeMetric.update(delta)
                   }
                 } catch {
-                  case _: ClosedByInterruptException =>
-                  case e: IOException =>
-                    task.notifier.setException(e)
-                    processIOException(e, DiskStatus.READ_OR_WRITE_FAILURE)
+                  case t: Throwable =>
+                    if (t.isInstanceOf[IOException]) {
+                      task.notifier.setException(t.asInstanceOf[IOException])
+                      processIOException(
+                        t.asInstanceOf[IOException],
+                        DiskStatus.READ_OR_WRITE_FAILURE)
+                    }
+                    logWarning(s"Flusher-${this}-thread-${index} encounter 
exception.", t)
                 }
                 lastBeginFlushTime.set(index, -1)
               }
-              returnBuffer(task.buffer)
+              Utils.tryLogNonFatalError(returnBuffer(task.buffer))
               task.notifier.numPendingFlushes.decrementAndGet()
             }
           }