This is an automated email from the ASF dual-hosted git repository.
ethanfeng 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 c3b129da1 [CELEBORN-1252][FOLLOWUP] Fix
Worker#computeResourceConsumption NullPointerException for
userResourceConsumption that does not contain given userIdentifier
c3b129da1 is described below
commit c3b129da1f06a615320b8ed897f5970c646de382
Author: SteNicholas <[email protected]>
AuthorDate: Mon Feb 5 19:55:01 2024 +0800
[CELEBORN-1252][FOLLOWUP] Fix Worker#computeResourceConsumption
NullPointerException for userResourceConsumption that does not contain given
userIdentifier
### What changes were proposed in this pull request?
Fix `Worker#computeResourceConsumption` `NullPointerException` for
`userResourceConsumption` that does not contain given `userIdentifier`.
### Why are the changes needed?
When `userResourceConsumption` of `workerInfo` does not contain given
`userIdentifier`, `Worker#computeResourceConsumption` causes
`NullPointerException` for worker dimension resource consumption metrics.
```
24/02/05 17:36:15,983 ERROR [worker-forward-message-scheduler] Utils:
Uncaught exception in thread worker-forward-message-scheduler
java.lang.NullPointerException
at
org.apache.celeborn.service.deploy.worker.Worker.$anonfun$gaugeResourceConsumption$1(Worker.scala:555)
at
scala.runtime.java8.JFunction0$mcJ$sp.apply(JFunction0$mcJ$sp.java:23)
at
org.apache.celeborn.common.metrics.source.GaugeSupplier$$anon$3.getValue(AbstractSource.scala:453)
at
org.apache.celeborn.common.metrics.source.AbstractSource.addGauge(AbstractSource.scala:79)
at
org.apache.celeborn.common.metrics.source.AbstractSource.addGauge(AbstractSource.scala:99)
at
org.apache.celeborn.service.deploy.worker.Worker.gaugeResourceConsumption(Worker.scala:554)
at
org.apache.celeborn.service.deploy.worker.Worker.$anonfun$handleResourceConsumption$1(Worker.scala:537)
at
org.apache.celeborn.service.deploy.worker.Worker.$anonfun$handleResourceConsumption$1$adapted(Worker.scala:536)
at scala.collection.immutable.Map$Map1.foreach(Map.scala:128)
at
org.apache.celeborn.service.deploy.worker.Worker.handleResourceConsumption(Worker.scala:536)
at
org.apache.celeborn.service.deploy.worker.Worker.org$apache$celeborn$service$deploy$worker$Worker$$heartbeatToMaster(Worker.scala:362)
at
org.apache.celeborn.service.deploy.worker.Worker$$anon$1.$anonfun$run$1(Worker.scala:395)
at
org.apache.celeborn.common.util.Utils$.tryLogNonFatalError(Utils.scala:230)
at
org.apache.celeborn.service.deploy.worker.Worker$$anon$1.run(Worker.scala:395)
at
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
```
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
GA and cluster.
Closes #2288 from SteNicholas/CELEBORN-1252.
Authored-by: SteNicholas <[email protected]>
Signed-off-by: mingji <[email protected]>
---
.../apache/celeborn/service/deploy/worker/Worker.scala | 17 ++++++++++-------
1 file changed, 10 insertions(+), 7 deletions(-)
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 30a0af7c5..46ff96f4f 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
@@ -533,6 +533,8 @@ private[celeborn] class Worker(
private def handleResourceConsumption(): util.Map[UserIdentifier,
ResourceConsumption] = {
val resourceConsumptionSnapshot =
storageManager.userResourceConsumptionSnapshot()
+ val userResourceConsumptions =
+
workerInfo.updateThenGetUserResourceConsumption(resourceConsumptionSnapshot.asJava)
resourceConsumptionSnapshot.foreach { case (userIdentifier,
userResourceConsumption) =>
gaugeResourceConsumption(userIdentifier)
val subResourceConsumptions =
userResourceConsumption.subResourceConsumptions
@@ -540,7 +542,7 @@ private[celeborn] class Worker(
subResourceConsumptions.asScala.keys.foreach {
gaugeResourceConsumption(userIdentifier, _) }
}
}
-
workerInfo.updateThenGetUserResourceConsumption(resourceConsumptionSnapshot.asJava)
+ userResourceConsumptions
}
private def gaugeResourceConsumption(
@@ -574,14 +576,15 @@ private[celeborn] class Worker(
private def computeResourceConsumption(
userIdentifier: UserIdentifier,
applicationId: String = null): ResourceConsumption = {
- var resourceConsumption =
workerInfo.userResourceConsumption.get(userIdentifier)
+ var resourceConsumption =
+ workerInfo.userResourceConsumption.getOrDefault(
+ userIdentifier,
+ ResourceConsumption(0, 0, 0, 0))
if (applicationId != null) {
val subResourceConsumptions = resourceConsumption.subResourceConsumptions
- if (CollectionUtils.isNotEmpty(subResourceConsumptions)
- && subResourceConsumptions.containsKey(applicationId)) {
- resourceConsumption = subResourceConsumptions.get(applicationId)
- } else {
- resourceConsumption = ResourceConsumption(0, 0, 0, 0)
+ if (CollectionUtils.isNotEmpty(subResourceConsumptions)) {
+ resourceConsumption =
+ subResourceConsumptions.getOrDefault(applicationId,
ResourceConsumption(0, 0, 0, 0))
}
}
resourceConsumption
