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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
