[incubator-celeborn] branch main updated: [CELEBORN-585] Create if not exists worker recoverPath when graceful shutdown is enabled (#1487)

[re...@apache.org]

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

rexxiong 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 65cdb3eba [CELEBORN-585] Create if not exists worker recoverPath when graceful shutdown is enabled (#1487)
65cdb3eba is described below

commit 65cdb3eba47dc02ba9e6d69bc0b7727a8e5e55b8
Author: Leo Li <ly...@163.com>
AuthorDate: Wed May 17 11:29:09 2023 +0800

    [CELEBORN-585] Create if not exists worker recoverPath when graceful shutdown is enabled (#1487)
---
 .../org/apache/celeborn/service/deploy/worker/Worker.scala | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
index 5318f7416..ec6a6c9fe 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
@@ -17,6 +17,7 @@
 
 package org.apache.celeborn.service.deploy.worker
 
+import java.io.File
 import java.lang.{Long => JLong}
 import java.util.{HashMap => JHashMap, HashSet => JHashSet}
 import java.util.concurrent._
@@ -79,6 +80,19 @@ private[celeborn] class Worker(
       conf.workerRpcPort != 0 && conf.workerFetchPort != 0 &&
       conf.workerPushPort != 0 && conf.workerReplicatePort != 0),
     "If enable graceful shutdown, the worker should use stable server port.")
+  if (gracefulShutdown) {
+    try {
+      val recoverRoot = new File(conf.workerRecoverPath)
+      if (!recoverRoot.exists()) {
+        logInfo(s"Recover root path ${conf.workerRecoverPath} does not exists, create it first.")
+        recoverRoot.mkdirs()
+      }
+    } catch {
+      case e: Exception =>
+        logError("Check or create recover root path failed: ", e)
+        throw e
+    }
+  }
 
   val workerSource = new WorkerSource(conf)
   metricsSystem.registerSource(workerSource)