This is an automated email from the ASF dual-hosted git repository.
rexxiong 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 fd7d28621 [CELEBORN-863][FOLLOWUP] Fix persisted committed file infos
lost
fd7d28621 is described below
commit fd7d28621f0aeaeca80f055d295e1145e84ad1f5
Author: mingji <[email protected]>
AuthorDate: Thu Mar 14 13:51:10 2024 +0800
[CELEBORN-863][FOLLOWUP] Fix persisted committed file infos lost
To fix a bug that might cause persisted committed file info lost.
A worker starts will clean its persisted committed file info and won't put
back if this worker restart again, the committed file infos will lost.
NO.
GA.
Closes #2390 from FMX/b863-1.
Authored-by: mingji <[email protected]>
Signed-off-by: Shuang <[email protected]>
(cherry picked from commit 5043f46aa737a64248230dd5301661657c0541fc)
Signed-off-by: Shuang <[email protected]>
---
.../celeborn/service/deploy/worker/storage/StorageManager.scala | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
index 624925c1b..597f64aa0 100644
---
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
+++
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
@@ -203,7 +203,8 @@ final private[worker] class StorageManager(conf:
CelebornConf, workerSource: Abs
conf.workerGracefulShutdownSaveCommittedFileInfoSync
private val saveCommittedFileInfoInterval =
conf.workerGracefulShutdownSaveCommittedFileInfoInterval
- private var committedFileInfos: ConcurrentHashMap[String,
ConcurrentHashMap[String, FileInfo]] = _
+ private val committedFileInfos: ConcurrentHashMap[String,
ConcurrentHashMap[String, FileInfo]] =
+ JavaUtils.newConcurrentHashMap[String, ConcurrentHashMap[String,
FileInfo]]()
// ShuffleClient can fetch data from a restarted worker only
// when the worker's fetching port is stable.
val workerGracefulShutdown = conf.workerGracefulShutdown
@@ -219,8 +220,6 @@ final private[worker] class StorageManager(conf:
CelebornConf, workerSource: Abs
logError("Init level DB failed:", e)
this.db = null
}
- committedFileInfos =
- JavaUtils.newConcurrentHashMap[String, ConcurrentHashMap[String,
FileInfo]]()
saveCommittedFileInfosExecutor =
ThreadUtils.newDaemonSingleThreadScheduledExecutor(
"worker-storage-manager-committed-fileinfo-saver")
@@ -266,6 +265,7 @@ final private[worker] class StorageManager(conf:
CelebornConf, workerSource: Abs
val files = PbSerDeUtils.fromPbFileInfoMap(entry.getValue, cache)
logDebug(s"Reload DB: $shuffleKey -> $files")
fileInfos.put(shuffleKey, files)
+ committedFileInfos.put(shuffleKey, files)
db.delete(entry.getKey)
} catch {
case exception: Exception =>
