Skip to content

[CELEBORN-2310] Reject RESERVE_SLOTS when disks are full #3666

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
saurabhd336 wants to merge 8 commits into from
+64 −4
Closed

[CELEBORN-2310] Reject RESERVE_SLOTS when disks are full #3666

Changes from 2 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
1c813ee
Reject RESERVER_SLOTS when disks are full
saurabhd336 Apr 16, 2026
625899d
Fix check
saurabhd336 Apr 16, 2026
c37148a
Review comments
saurabhd336 Apr 20, 2026
ad5ad32
Fix test
saurabhd336 Apr 20, 2026
38803c2
Fix test
saurabhd336 Apr 20, 2026
7f7ffac
Enhance logs
saurabhd336 Apr 20, 2026
098a2ed
Merge branch 'main' of https://github.com/apache/celeborn into diskFu…
saurabhd336 May 14, 2026
896bfda
Fix test
saurabhd336 May 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,10 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
}

def healthyWorkingDirs(): List[File] =
disksSnapshot().filter(_.status == DiskStatus.HEALTHY).flatMap(_.dirs)
disksSnapshot()
.filter(diskInfo =>
(diskInfo.status == DiskStatus.HEALTHY) && (diskInfo.actualUsableSpace > 0))
.flatMap(_.dirs)
Comment thread
SteNicholas marked this conversation as resolved.
Outdated
Comment thread
SteNicholas marked this conversation as resolved.
Outdated

private val diskOperators: ConcurrentHashMap[String, ThreadPoolExecutor] = {
val cleaners = JavaUtils.newConcurrentHashMap[String, ThreadPoolExecutor]()
Expand Down Expand Up @@ -1139,7 +1142,8 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
while (retryCount < conf.workerCreateWriterMaxAttempts) {
val diskInfo = diskInfos.get(suggestedMountPoint)
val dirs =
if (diskInfo != null && diskInfo.status.equals(DiskStatus.HEALTHY)) {
if (diskInfo != null && diskInfo.status.equals(
DiskStatus.HEALTHY) && diskInfo.actualUsableSpace > 0) {
diskInfo.dirs
} else {
if (suggestedMountPoint.isEmpty) {
Comment thread
zaynt4606 marked this conversation as resolved.
Expand Down
Loading