You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "Dzeri96 (via GitHub)" <gi...@apache.org> on 2026/02/06 15:04:56 UTC
[PR] [CELEBORN-2257] Fixed remote disks not being reported on registration [celeborn]
Dzeri96 opened a new pull request, #3597:
URL: https://github.com/apache/celeborn/pull/3597
<!--
Thanks for sending a pull request! Here are some tips for you:
- Make sure the PR title start w/ a JIRA ticket, e.g. '[CELEBORN-XXXX] Your PR title ...'.
- Be sure to keep the PR description updated to reflect all changes.
- Please write your PR title to summarize what this PR proposes.
- If possible, provide a concise example to reproduce the issue for a faster review.
-->
### What changes were proposed in this pull request?
1. Disks reported to the master on registration now include remote disks (HDFS, S3, OSS)
2. Refactored method names to clarify difference between local and remote disks.
3. Embedded disk type information into the enum.
4. Refactored unnecessarily complicated code in the slot assignment and worker registration path.
### Why are the changes needed?
1. Before the first heartbeat, the master won't be able to assign slots from the remote disks on the worker.
2. All other changes are in preparation for better support of remote disks.
### Does this PR resolve a correctness bug?
Yes
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
**Important**: I want help from the community on how to write tests for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on code in PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#discussion_r2811698980
##########
common/src/main/java/org/apache/celeborn/common/protocol/StorageInfo.java:
##########
@@ -22,22 +22,34 @@
public class StorageInfo implements Serializable {
public enum Type {
- MEMORY(0),
- HDD(1),
- SSD(2),
- HDFS(3),
- OSS(4),
- S3(5);
+ MEMORY(0, false, MEMORY_MASK),
+ HDD(1, false, LOCAL_DISK_MASK),
+ SSD(2, false, LOCAL_DISK_MASK),
+ HDFS(3, true, HDFS_MASK),
+ OSS(4, true, OSS_MASK),
+ S3(5, true, S3_MASK);
private final int value;
+ private final boolean isDFS;
+ private final int mask;
- Type(int value) {
+ Type(int value, boolean isDFS, int mask) {
this.value = value;
+ this.isDFS = isDFS;
Review Comment:
So my initial idea for implementing this was a HashMap. I had built a map that was being filled in the `static` block, like the other maps, but then I realized you had to make sure each ENUM member was in this map, so I wrote a test to enforce it.
In the end I found this static solution much more elegant. The compiler forces you to assign each enum member a isDFS value. It's also less code. Let me know if you want me to change it though.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on code in PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#discussion_r2787051312
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala:
##########
@@ -284,11 +284,10 @@ private[celeborn] class Worker(
storageManager.updateDiskInfos()
storageManager.startDeviceMonitor()
- // WorkerInfo's diskInfos is a reference to storageManager.diskInfos
- val diskInfos = JavaUtils.newConcurrentHashMap[String, DiskInfo]()
- storageManager.disksSnapshot().foreach { diskInfo =>
- diskInfos.put(diskInfo.mountPoint, diskInfo)
- }
+ private val diskInfos = storageManager
+ .allDisksSnapshot()
+ .map { diskInfo => diskInfo.mountPoint -> diskInfo }
+ .toMap.asJava
Review Comment:
Furthermore, if this were true, the right solution would be to encapsulate this concern inside of the `WorkerInfo` class, which is exactly what we are doing now. Copilot is promoting something that goes against object-oriented principles.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Add reporting of remote disks during registration. [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#issuecomment-4440459805
@SteNicholas Actually, the tests seem flaky. Sometimes they pass and sometimes they don't. Do you think this has something to do with my code or was it there before?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on code in PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#discussion_r2779142325
##########
master/src/main/java/org/apache/celeborn/service/deploy/master/SlotsAllocator.java:
##########
@@ -157,9 +144,7 @@ static class UsableDiskInfo {
diskToWorkerMap.put(diskInfo, i);
if (diskInfo.actualUsableSpace() > 0
&& diskInfo.status().equals(DiskStatus.HEALTHY)
- && diskInfo.storageType() != StorageInfo.Type.HDFS
- && diskInfo.storageType() != StorageInfo.Type.S3
- && diskInfo.storageType() != StorageInfo.Type.OSS) {
+ && !diskInfo.storageType().isDFS()) {
usableDisks.add(diskInfo);
}
}));
Review Comment:
True. Same feedback as above. Will revert for now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Add reporting of remote disks during registration. [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#issuecomment-4443140482
@Dzeri96, you could fix the flaky tests in another pull request. I would like to rerun the failed tests until succeed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Add reporting of remote disks during registration. [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas closed pull request #3597: [CELEBORN-2257] Add reporting of remote disks during registration.
URL: https://github.com/apache/celeborn/pull/3597
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#discussion_r2796501849
##########
common/src/main/scala/org/apache/celeborn/common/meta/WorkerInfo.scala:
##########
@@ -214,25 +214,21 @@ class WorkerInfo(
for (newDisk <- newDiskInfos.values().asScala) {
val mountPoint: String = newDisk.mountPoint
val curDisk = diskInfos.get(mountPoint)
+ if (estimatedPartitionSize.nonEmpty && !newDisk.storageType.isDFS) {
+ newDisk.maxSlots = newDisk.totalSpace / estimatedPartitionSize.get
+ newDisk.availableSlots = newDisk.actualUsableSpace / estimatedPartitionSize.get
+ }
Review Comment:
This change introduces new slot-update semantics for DFS vs local disks inside `updateThenGetDiskInfos`, but there are no unit tests covering that remote/DFS disk slot fields are preserved across successive updates (e.g., registration `updateDiskSlots(...)` followed by heartbeat `updateThenGetDiskInfos(...)`). Adding a focused test in `WorkerInfoSuite` for a DFS `DiskInfo` would help prevent regressions like remote disks becoming unavailable after the first heartbeat.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Add reporting of remote disks during registration. [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#issuecomment-4448844386
Thank you and sorry for the drawn-out PR. I won't submit anything else before fixing the flaky tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Add reporting of remote disks during registration. [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#issuecomment-4303998013
The logs have expired in the meantime, but I think that the failing tests are not because of me. You guys can try to re-trigger them and take a look.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#discussion_r2785653797
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala:
##########
@@ -284,11 +284,10 @@ private[celeborn] class Worker(
storageManager.updateDiskInfos()
storageManager.startDeviceMonitor()
- // WorkerInfo's diskInfos is a reference to storageManager.diskInfos
- val diskInfos = JavaUtils.newConcurrentHashMap[String, DiskInfo]()
- storageManager.disksSnapshot().foreach { diskInfo =>
- diskInfos.put(diskInfo.mountPoint, diskInfo)
- }
+ private val diskInfos = storageManager
+ .allDisksSnapshot()
+ .map { diskInfo => diskInfo.mountPoint -> diskInfo }
+ .toMap.asJava
Review Comment:
`diskInfos` is initialized with `.toMap.asJava`, which produces a Java view over an *immutable* Scala `Map`. `WorkerInfo` mutates the passed-in map later (e.g., `diskInfos.put/remove` in `updateThenGetDiskInfos`), so this can throw `UnsupportedOperationException` at runtime. Initialize `diskInfos` as a mutable `ConcurrentHashMap` (e.g., via `JavaUtils.newConcurrentHashMap` and `put`s, or by copying into a `ConcurrentHashMap`) before passing it to `WorkerInfo`.
```suggestion
private val diskInfos: JMap[String, DiskInfo] = {
val map = JavaUtils.newConcurrentHashMap[String, DiskInfo]
storageManager
.allDisksSnapshot()
.foreach { diskInfo =>
map.put(diskInfo.mountPoint, diskInfo)
}
map
}
```
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala:
##########
@@ -284,11 +284,10 @@ private[celeborn] class Worker(
storageManager.updateDiskInfos()
storageManager.startDeviceMonitor()
- // WorkerInfo's diskInfos is a reference to storageManager.diskInfos
- val diskInfos = JavaUtils.newConcurrentHashMap[String, DiskInfo]()
- storageManager.disksSnapshot().foreach { diskInfo =>
- diskInfos.put(diskInfo.mountPoint, diskInfo)
- }
+ private val diskInfos = storageManager
+ .allDisksSnapshot()
+ .map { diskInfo => diskInfo.mountPoint -> diskInfo }
+ .toMap.asJava
Review Comment:
This PR changes worker registration/heartbeat disk reporting to include remote disks (`allDisksSnapshot`) and introduces new slot-availability semantics (`StorageInfo.isAvailable`, `Type.isDFS`). There doesn’t appear to be a test asserting that remote disk infos are (a) included in the initial registration payload and (b) preserved across subsequent heartbeats so the master can allocate slots from them before/without the first heartbeat. Adding a focused unit/integration test around worker->master disk info propagation would help prevent regressions here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Add reporting of remote disks during registration. [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#issuecomment-4446732589
Thanks. Merged to main(v0.7.0).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Add reporting of remote disks during registration. [celeborn]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#issuecomment-4168528286
This PR is stale because it has been open 20 days with no activity. Remove stale label or comment or this will be closed in 10 days.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on code in PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#discussion_r2779120047
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala:
##########
@@ -102,19 +102,23 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
if (diskInfoSet.nonEmpty) Some(diskInfoSet) else None
}
- def disksSnapshot(): List[DiskInfo] = {
+ def localDisksSnapshot(): List[DiskInfo] = {
diskInfos.synchronized {
val disks = new util.ArrayList[DiskInfo](diskInfos.values())
disks.asScala.toList
}
}
- def healthyWorkingDirs(): List[File] =
- disksSnapshot().filter(_.status == DiskStatus.HEALTHY).flatMap(_.dirs)
+ def allDisksSnapshot(): List[DiskInfo] = {
+ localDisksSnapshot() ++ remoteDiskInfos.getOrElse(Nil)
Review Comment:
This is straight up incorrect. First of all, the code compiles and the tests pass. Second of all, this can be verified easily in the scala shell:
```scala
Welcome to Scala 2.13.17 (OpenJDK 64-Bit Server VM, Java 17.0.16).
Type in expressions for evaluation. Or try :help.
scala> List(1,2,3)
val res0: List[Int] = List(1, 2, 3)
scala> val list = List(1,2,3)
val list: List[Int] = List(1, 2, 3)
scala> Option(Set(5,6))
val res1: Option[scala.collection.immutable.Set[Int]] = Some(Set(5, 6))
scala> val optSome = Option(Set(5,6))
val optSome: Option[scala.collection.immutable.Set[Int]] = Some(Set(5, 6))
scala> val optNone = Option.empty
val optNone: Option[Nothing] = None
scala> list ++ optSome.getOrElse(Nil)
val res2: List[Int] = List(1, 2, 3, 5, 6)
scala> list ++ optNone.getOrElse(Nil)
val res3: List[Int] = List(1, 2, 3)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#discussion_r2796482941
##########
common/src/main/java/org/apache/celeborn/common/protocol/StorageInfo.java:
##########
@@ -22,22 +22,34 @@
public class StorageInfo implements Serializable {
public enum Type {
- MEMORY(0),
- HDD(1),
- SSD(2),
- HDFS(3),
- OSS(4),
- S3(5);
+ MEMORY(0, false, MEMORY_MASK),
+ HDD(1, false, LOCAL_DISK_MASK),
+ SSD(2, false, LOCAL_DISK_MASK),
+ HDFS(3, true, HDFS_MASK),
+ OSS(4, true, OSS_MASK),
+ S3(5, true, S3_MASK);
private final int value;
+ private final boolean isDFS;
+ private final int mask;
- Type(int value) {
+ Type(int value, boolean isDFS, int mask) {
this.value = value;
+ this.isDFS = isDFS;
Review Comment:
IMO, it's unnecessary to add `isDFS` variable. The `isDFS` method is enough for usage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on code in PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#discussion_r2787039684
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala:
##########
@@ -284,11 +284,10 @@ private[celeborn] class Worker(
storageManager.updateDiskInfos()
storageManager.startDeviceMonitor()
- // WorkerInfo's diskInfos is a reference to storageManager.diskInfos
- val diskInfos = JavaUtils.newConcurrentHashMap[String, DiskInfo]()
- storageManager.disksSnapshot().foreach { diskInfo =>
- diskInfos.put(diskInfo.mountPoint, diskInfo)
- }
+ private val diskInfos = storageManager
+ .allDisksSnapshot()
+ .map { diskInfo => diskInfo.mountPoint -> diskInfo }
+ .toMap.asJava
Review Comment:
See... this is straight up wrong. In WorkerInfo's constructor, the argument is copied into a `ConcurrentHashMap` before being assigned to a local member:
```scala
val diskInfos = {
if (_diskInfos != null) JavaUtils.newConcurrentHashMap[String, DiskInfo](_diskInfos)
else null
}
```
I agree that AI can spot some issues that are impossible for compilers, and very hard for humans to catch, but half the time you spend arguing with it over obviously wrong and easily-verifiable statements. I think it's much better for a human to take an hour to review this, than to spend 3 hours arguing with an AI.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#issuecomment-3907687232
@SteNicholas So my changes are explained in the PR description pretty well I think. While @eolivelli was running his tests, he noticed that the current faulty behaviour presents a problem when the auto-scaling spawns new nodes. In this moment, the system is under pressure, and yet the newly-spawned nodes don't report remote disks, leading to performance degredation and the need to spawn more nodes.
In hindsight though, I should have limited the PR to just this. It's just that while I was trying to understand the code, I made the other changes to make it more readable for myself. In the end I decided to include them too since we will be working on this part of the project a lot in the future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Add reporting of remote disks during registration. [celeborn]
Posted by "eolivelli (via GitHub)" <gi...@apache.org>.
eolivelli commented on PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#issuecomment-4371533397
Please check and fix CI
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Add reporting of remote disks during registration. [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#discussion_r2852150612
##########
master/src/main/java/org/apache/celeborn/service/deploy/master/SlotsAllocator.java:
##########
@@ -70,31 +78,10 @@ static class UsableDiskInfo {
for (WorkerInfo worker : workers) {
List<UsableDiskInfo> usableDisks =
slotsRestrictions.computeIfAbsent(worker, v -> new ArrayList<>());
- for (Map.Entry<String, DiskInfo> diskInfoEntry : worker.diskInfos().entrySet()) {
- if (diskInfoEntry.getValue().status().equals(DiskStatus.HEALTHY)) {
- if (StorageInfo.localDiskAvailable(availableStorageTypes)
- && diskInfoEntry.getValue().storageType() != StorageInfo.Type.HDFS
- && diskInfoEntry.getValue().storageType() != StorageInfo.Type.S3
- && diskInfoEntry.getValue().storageType() != StorageInfo.Type.OSS) {
- usableDisks.add(
- new UsableDiskInfo(
- diskInfoEntry.getValue(), diskInfoEntry.getValue().getAvailableSlots()));
- } else if (StorageInfo.HDFSAvailable(availableStorageTypes)
- && diskInfoEntry.getValue().storageType() == StorageInfo.Type.HDFS) {
- usableDisks.add(
- new UsableDiskInfo(
- diskInfoEntry.getValue(), diskInfoEntry.getValue().getAvailableSlots()));
- } else if (StorageInfo.S3Available(availableStorageTypes)
- && diskInfoEntry.getValue().storageType() == StorageInfo.Type.S3) {
- usableDisks.add(
- new UsableDiskInfo(
- diskInfoEntry.getValue(), diskInfoEntry.getValue().getAvailableSlots()));
- } else if (StorageInfo.OSSAvailable(availableStorageTypes)
- && diskInfoEntry.getValue().storageType() == StorageInfo.Type.OSS) {
- usableDisks.add(
- new UsableDiskInfo(
- diskInfoEntry.getValue(), diskInfoEntry.getValue().availableSlots()));
- }
+ for (DiskInfo diskInfo : worker.diskInfos().values()) {
+ if (diskInfo.status().equals(DiskStatus.HEALTHY)
Review Comment:
```suggestion
if (DiskStatus.HEALTHY.equals(diskInfo.status())
```
This suggestion could avoid the null situation for `diskInfo.status()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#discussion_r2778562928
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala:
##########
@@ -514,10 +513,10 @@ private[celeborn] class Worker(
activeShuffleKeys.addAll(partitionLocationInfo.shuffleKeySet)
activeShuffleKeys.addAll(storageManager.shuffleKeySet())
storageManager.updateDiskInfos()
- val diskInfos =
- workerInfo.updateThenGetDiskInfos(storageManager.disksSnapshot().map { disk =>
- disk.mountPoint -> disk
- }.toMap.asJava).values().asScala.toSeq ++ storageManager.remoteDiskInfos.getOrElse(Set.empty)
+ val currentDiskMap = storageManager.localDisksSnapshot().map { disk =>
Review Comment:
Heartbeat updates only pass `localDisksSnapshot()` into `workerInfo.updateThenGetDiskInfos(...)`. Since `updateThenGetDiskInfos` removes mount points missing from the update map, this will drop the remote disk entries that were added at registration, so remote disks stop being reported after the first heartbeat. Build the update map from `allDisksSnapshot()` (or merge in `remoteDiskInfos`) so remote disks remain present.
```suggestion
val currentDiskMap = storageManager.allDisksSnapshot().map { disk =>
```
##########
master/src/main/java/org/apache/celeborn/service/deploy/master/SlotsAllocator.java:
##########
@@ -69,32 +77,11 @@ static class UsableDiskInfo {
Map<WorkerInfo, List<UsableDiskInfo>> slotsRestrictions = new HashMap<>();
for (WorkerInfo worker : workers) {
List<UsableDiskInfo> usableDisks =
- slotsRestrictions.computeIfAbsent(worker, v -> new ArrayList<>());
- for (Map.Entry<String, DiskInfo> diskInfoEntry : worker.diskInfos().entrySet()) {
- if (diskInfoEntry.getValue().status().equals(DiskStatus.HEALTHY)) {
- if (StorageInfo.localDiskAvailable(availableStorageTypes)
- && diskInfoEntry.getValue().storageType() != StorageInfo.Type.HDFS
- && diskInfoEntry.getValue().storageType() != StorageInfo.Type.S3
- && diskInfoEntry.getValue().storageType() != StorageInfo.Type.OSS) {
- usableDisks.add(
- new UsableDiskInfo(
- diskInfoEntry.getValue(), diskInfoEntry.getValue().getAvailableSlots()));
- } else if (StorageInfo.HDFSAvailable(availableStorageTypes)
- && diskInfoEntry.getValue().storageType() == StorageInfo.Type.HDFS) {
- usableDisks.add(
- new UsableDiskInfo(
- diskInfoEntry.getValue(), diskInfoEntry.getValue().getAvailableSlots()));
- } else if (StorageInfo.S3Available(availableStorageTypes)
- && diskInfoEntry.getValue().storageType() == StorageInfo.Type.S3) {
- usableDisks.add(
- new UsableDiskInfo(
- diskInfoEntry.getValue(), diskInfoEntry.getValue().getAvailableSlots()));
- } else if (StorageInfo.OSSAvailable(availableStorageTypes)
- && diskInfoEntry.getValue().storageType() == StorageInfo.Type.OSS) {
- usableDisks.add(
- new UsableDiskInfo(
- diskInfoEntry.getValue(), diskInfoEntry.getValue().availableSlots()));
- }
+ slotsRestrictions.computeIfAbsent(worker, v -> new LinkedList<>());
+ for (DiskInfo diskInfo : worker.diskInfos().values()) {
+ if (diskInfo.status().equals(DiskStatus.HEALTHY)
+ && StorageInfo.isAvailable(diskInfo.storageType(), availableStorageTypes)) {
+ usableDisks.add(new UsableDiskInfo(diskInfo));
}
Review Comment:
`usableDisks` is created as a `LinkedList`, but later code accesses it by index via `usableDiskInfos.get(diskIndex)` in `getStorageInfo(...)`. With `LinkedList` this becomes O(n) per access and can significantly slow slot allocation for large numbers of partitions/disks. Use an `ArrayList` for `usableDisks` here.
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala:
##########
@@ -102,19 +102,23 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
if (diskInfoSet.nonEmpty) Some(diskInfoSet) else None
}
- def disksSnapshot(): List[DiskInfo] = {
+ def localDisksSnapshot(): List[DiskInfo] = {
diskInfos.synchronized {
val disks = new util.ArrayList[DiskInfo](diskInfos.values())
disks.asScala.toList
}
}
- def healthyWorkingDirs(): List[File] =
- disksSnapshot().filter(_.status == DiskStatus.HEALTHY).flatMap(_.dirs)
+ def allDisksSnapshot(): List[DiskInfo] = {
+ localDisksSnapshot() ++ remoteDiskInfos.getOrElse(Nil)
Review Comment:
`allDisksSnapshot()` currently won’t compile: `remoteDiskInfos` is an `Option[Set[DiskInfo]]`, but `getOrElse(Nil)` returns a `List` (`Nil`). Use `getOrElse(Set.empty)` (and optionally `.toList`) or change `remoteDiskInfos` to a sequence type so the concatenation is type-safe.
```suggestion
localDisksSnapshot() ++ remoteDiskInfos.getOrElse(Set.empty[DiskInfo]).toList
```
##########
common/src/main/scala/org/apache/celeborn/common/meta/WorkerInfo.scala:
##########
@@ -214,25 +214,21 @@ class WorkerInfo(
for (newDisk <- newDiskInfos.values().asScala) {
val mountPoint: String = newDisk.mountPoint
val curDisk = diskInfos.get(mountPoint)
+ if (estimatedPartitionSize.nonEmpty && !newDisk.storageType.isDFS) {
+ newDisk.maxSlots = newDisk.totalSpace / estimatedPartitionSize.get
+ newDisk.availableSlots = newDisk.actualUsableSpace / estimatedPartitionSize.get
+ }
if (curDisk != null) {
curDisk.actualUsableSpace = newDisk.actualUsableSpace
curDisk.totalSpace = newDisk.totalSpace
// Update master's diskinfo activeslots to worker's value
curDisk.activeSlots = newDisk.activeSlots
curDisk.avgFlushTime = newDisk.avgFlushTime
curDisk.avgFetchTime = newDisk.avgFetchTime
- if (estimatedPartitionSize.nonEmpty && curDisk.storageType != StorageInfo.Type.HDFS
- && curDisk.storageType != StorageInfo.Type.S3 && curDisk.storageType != StorageInfo.Type.OSS) {
- curDisk.maxSlots = curDisk.totalSpace / estimatedPartitionSize.get
- curDisk.availableSlots = curDisk.actualUsableSpace / estimatedPartitionSize.get
- }
+ curDisk.maxSlots = newDisk.maxSlots
+ curDisk.availableSlots = newDisk.availableSlots
Review Comment:
`updateThenGetDiskInfos` now copies `newDisk.maxSlots/availableSlots` into the existing `curDisk` for all disk types. For DFS/remote disks, the worker-side `DiskInfo` often has `maxSlots/availableSlots` left at default (0), while the master computes usable slots during registration via `updateDiskSlots(...)`. This change can therefore zero out the master-computed slots for remote disks on the first heartbeat, making them unusable. Consider recomputing slots on the master for DFS disks as well, or avoid overwriting `curDisk.maxSlots/availableSlots` for `storageType.isDFS` disks unless the worker provides meaningful values.
```suggestion
// Only overwrite slot-related fields when worker reports meaningful values.
if (newDisk.maxSlots > 0) {
curDisk.maxSlots = newDisk.maxSlots
}
if (newDisk.availableSlots > 0) {
curDisk.availableSlots = newDisk.availableSlots
}
```
##########
master/src/main/java/org/apache/celeborn/service/deploy/master/SlotsAllocator.java:
##########
@@ -157,9 +144,7 @@ static class UsableDiskInfo {
diskToWorkerMap.put(diskInfo, i);
if (diskInfo.actualUsableSpace() > 0
&& diskInfo.status().equals(DiskStatus.HEALTHY)
- && diskInfo.storageType() != StorageInfo.Type.HDFS
- && diskInfo.storageType() != StorageInfo.Type.S3
- && diskInfo.storageType() != StorageInfo.Type.OSS) {
+ && !diskInfo.storageType().isDFS()) {
usableDisks.add(diskInfo);
}
}));
Review Comment:
`usableDisks` was switched to `LinkedList`, but `placeDisksToGroups(...)` sorts it and repeatedly uses `subList(start, end)`. With `LinkedList`, these operations have extra traversal overhead compared to `ArrayList`. Unless there’s a specific need for O(1) inserts in the middle (not apparent here), prefer `ArrayList` for better performance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on code in PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#discussion_r2779139627
##########
master/src/main/java/org/apache/celeborn/service/deploy/master/SlotsAllocator.java:
##########
@@ -69,32 +77,11 @@ static class UsableDiskInfo {
Map<WorkerInfo, List<UsableDiskInfo>> slotsRestrictions = new HashMap<>();
for (WorkerInfo worker : workers) {
List<UsableDiskInfo> usableDisks =
- slotsRestrictions.computeIfAbsent(worker, v -> new ArrayList<>());
- for (Map.Entry<String, DiskInfo> diskInfoEntry : worker.diskInfos().entrySet()) {
- if (diskInfoEntry.getValue().status().equals(DiskStatus.HEALTHY)) {
- if (StorageInfo.localDiskAvailable(availableStorageTypes)
- && diskInfoEntry.getValue().storageType() != StorageInfo.Type.HDFS
- && diskInfoEntry.getValue().storageType() != StorageInfo.Type.S3
- && diskInfoEntry.getValue().storageType() != StorageInfo.Type.OSS) {
- usableDisks.add(
- new UsableDiskInfo(
- diskInfoEntry.getValue(), diskInfoEntry.getValue().getAvailableSlots()));
- } else if (StorageInfo.HDFSAvailable(availableStorageTypes)
- && diskInfoEntry.getValue().storageType() == StorageInfo.Type.HDFS) {
- usableDisks.add(
- new UsableDiskInfo(
- diskInfoEntry.getValue(), diskInfoEntry.getValue().getAvailableSlots()));
- } else if (StorageInfo.S3Available(availableStorageTypes)
- && diskInfoEntry.getValue().storageType() == StorageInfo.Type.S3) {
- usableDisks.add(
- new UsableDiskInfo(
- diskInfoEntry.getValue(), diskInfoEntry.getValue().getAvailableSlots()));
- } else if (StorageInfo.OSSAvailable(availableStorageTypes)
- && diskInfoEntry.getValue().storageType() == StorageInfo.Type.OSS) {
- usableDisks.add(
- new UsableDiskInfo(
- diskInfoEntry.getValue(), diskInfoEntry.getValue().availableSlots()));
- }
+ slotsRestrictions.computeIfAbsent(worker, v -> new LinkedList<>());
+ for (DiskInfo diskInfo : worker.diskInfos().values()) {
+ if (diskInfo.status().equals(DiskStatus.HEALTHY)
+ && StorageInfo.isAvailable(diskInfo.storageType(), availableStorageTypes)) {
+ usableDisks.add(new UsableDiskInfo(diskInfo));
}
Review Comment:
True, however the place where this list is actually being used is three method calls deep. We should maybe refactor the code to restrict this type of dependency.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on code in PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#discussion_r2779130269
##########
common/src/main/scala/org/apache/celeborn/common/meta/WorkerInfo.scala:
##########
@@ -214,25 +214,21 @@ class WorkerInfo(
for (newDisk <- newDiskInfos.values().asScala) {
val mountPoint: String = newDisk.mountPoint
val curDisk = diskInfos.get(mountPoint)
+ if (estimatedPartitionSize.nonEmpty && !newDisk.storageType.isDFS) {
+ newDisk.maxSlots = newDisk.totalSpace / estimatedPartitionSize.get
+ newDisk.availableSlots = newDisk.actualUsableSpace / estimatedPartitionSize.get
+ }
if (curDisk != null) {
curDisk.actualUsableSpace = newDisk.actualUsableSpace
curDisk.totalSpace = newDisk.totalSpace
// Update master's diskinfo activeslots to worker's value
curDisk.activeSlots = newDisk.activeSlots
curDisk.avgFlushTime = newDisk.avgFlushTime
curDisk.avgFetchTime = newDisk.avgFetchTime
- if (estimatedPartitionSize.nonEmpty && curDisk.storageType != StorageInfo.Type.HDFS
- && curDisk.storageType != StorageInfo.Type.S3 && curDisk.storageType != StorageInfo.Type.OSS) {
- curDisk.maxSlots = curDisk.totalSpace / estimatedPartitionSize.get
- curDisk.availableSlots = curDisk.actualUsableSpace / estimatedPartitionSize.get
- }
+ curDisk.maxSlots = newDisk.maxSlots
+ curDisk.availableSlots = newDisk.availableSlots
Review Comment:
This was the case before my change. We could fix it now I guess...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#issuecomment-3952773149
I'll try to write some tests this week and will ping you when I'm done @RexXiong
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Add reporting of remote disks during registration. [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#issuecomment-4168537407
I haven't forgotten about this PR. It's just that the tests have proven to be hard to write and I hadn't had that much time. Hopefully I'll do it soon!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong commented on PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#issuecomment-3949965716
> @SteNicholas So my changes are explained in the PR description pretty well I think. While @eolivelli was running his tests, he noticed that the current faulty behaviour presents a problem when the auto-scaling spawns new nodes. In this moment, the system is under pressure, and yet the newly-spawned nodes don't report remote disks, leading to performance degredation and the need to spawn more nodes.
>
> In hindsight though, I should have limited the PR to just this. It's just that while I was trying to understand the code, I made the other changes to make it more readable for myself. In the end I decided to include them too since we will be working on this part of the project a lot in the future.
>
> Also, don't forget to help me with writing a test!
for PR overall LGTM, Additionally, you can use `MiniClusterFeature` to create a virtual cluster and validate the worker's registration with the master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Add reporting of remote disks during registration. [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#issuecomment-4245387067
I've spent so much time fighting Mockito on this one ([issue](https://github.com/mockito/mockito/issues/3811) opened), that I forgot what I wrote in this PR.
Finally I managed to write a half-decent test that mocks the `MasterClient` and intercepts the communication between the two.
In the future, we should inject this client in the Master/Worker to make testing easier, but for now, this should work.
@SteNicholas Please be so kind to take another look.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Add reporting of remote disks during registration. [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#issuecomment-4439294700
I can locally reproduce some failing tests from the service suite. Let me check what's wrong...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#issuecomment-3866166147
## [Codecov](https://app.codecov.io/gh/apache/celeborn/pull/3597?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `94.44444%` with `1 line` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 67.07%. Comparing base ([`2dd1b7a`](https://app.codecov.io/gh/apache/celeborn/commit/2dd1b7aac4d488c0496da1fc77cab9c79bccd69b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`d5ae63a`](https://app.codecov.io/gh/apache/celeborn/commit/d5ae63ab87f8ac5d449c1f02511e6f668e6b3dd1?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 13 commits behind head on main.
| [Files with missing lines](https://app.codecov.io/gh/apache/celeborn/pull/3597?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...a/org/apache/celeborn/common/meta/WorkerInfo.scala](https://app.codecov.io/gh/apache/celeborn/pull/3597?src=pr&el=tree&filepath=common%2Fsrc%2Fmain%2Fscala%2Forg%2Fapache%2Fceleborn%2Fcommon%2Fmeta%2FWorkerInfo.scala&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvY2VsZWJvcm4vY29tbW9uL21ldGEvV29ya2VySW5mby5zY2FsYQ==) | 80.00% | [0 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/celeborn/pull/3597?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #3597 +/- ##
==========================================
- Coverage 67.13% 67.07% -0.06%
==========================================
Files 357 357
Lines 21860 21935 +75
Branches 1943 1947 +4
==========================================
+ Hits 14674 14711 +37
- Misses 6166 6213 +47
+ Partials 1020 1011 -9
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/celeborn/pull/3597?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on code in PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#discussion_r2786964590
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala:
##########
@@ -284,11 +284,10 @@ private[celeborn] class Worker(
storageManager.updateDiskInfos()
storageManager.startDeviceMonitor()
- // WorkerInfo's diskInfos is a reference to storageManager.diskInfos
- val diskInfos = JavaUtils.newConcurrentHashMap[String, DiskInfo]()
- storageManager.disksSnapshot().foreach { diskInfo =>
- diskInfos.put(diskInfo.mountPoint, diskInfo)
- }
+ private val diskInfos = storageManager
+ .allDisksSnapshot()
+ .map { diskInfo => diskInfo.mountPoint -> diskInfo }
+ .toMap.asJava
Review Comment:
This is what I wrote in the original PR. I need someone from the existing community to guide me on writing an integration test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2257] Fix remote disks not being reported on registration [celeborn]
Posted by "Dzeri96 (via GitHub)" <gi...@apache.org>.
Dzeri96 commented on code in PR #3597:
URL: https://github.com/apache/celeborn/pull/3597#discussion_r2779126662
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala:
##########
@@ -514,10 +513,10 @@ private[celeborn] class Worker(
activeShuffleKeys.addAll(partitionLocationInfo.shuffleKeySet)
activeShuffleKeys.addAll(storageManager.shuffleKeySet())
storageManager.updateDiskInfos()
- val diskInfos =
- workerInfo.updateThenGetDiskInfos(storageManager.disksSnapshot().map { disk =>
- disk.mountPoint -> disk
- }.toMap.asJava).values().asScala.toSeq ++ storageManager.remoteDiskInfos.getOrElse(Set.empty)
+ val currentDiskMap = storageManager.localDisksSnapshot().map { disk =>
Review Comment:
True, I forgot to change the method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org