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 277d060e4 [CELEBORN-1174][FOLLOWUP] Fix
Worker#computeResourceConsumption NullPointerException with null
subResourceConsumptions
277d060e4 is described below
commit 277d060e4895a83e3f83c4ea45a6b56264e1c8d1
Author: SteNicholas <[email protected]>
AuthorDate: Sun Feb 4 20:24:12 2024 +0800
[CELEBORN-1174][FOLLOWUP] Fix Worker#computeResourceConsumption
NullPointerException with null subResourceConsumptions
### What changes were proposed in this pull request?
Fix `Worker#computeResourceConsumption` `NullPointerException` with null
`subResourceConsumptions`.
### Why are the changes needed?
With null `subResourceConsumptions`, `Worker#computeResourceConsumption`
causes `NullPointerException` for application dimension resource consumption
metrics.
```
24/02/04 13:58:13,757 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.computeResourceConsumption(Worker.scala:581)
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 #2286 from SteNicholas/CELEBORN-1174.
Authored-by: SteNicholas <[email protected]>
Signed-off-by: mingji <[email protected]>
---
.../org/apache/celeborn/service/deploy/worker/Worker.scala | 10 +++++++---
1 file changed, 7 insertions(+), 3 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 b1b5a72cc..30a0af7c5 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
@@ -576,9 +576,13 @@ private[celeborn] class Worker(
applicationId: String = null): ResourceConsumption = {
var resourceConsumption =
workerInfo.userResourceConsumption.get(userIdentifier)
if (applicationId != null) {
- resourceConsumption =
resourceConsumption.subResourceConsumptions.getOrDefault(
- applicationId,
- ResourceConsumption(0, 0, 0, 0))
+ val subResourceConsumptions = resourceConsumption.subResourceConsumptions
+ if (CollectionUtils.isNotEmpty(subResourceConsumptions)
+ && subResourceConsumptions.containsKey(applicationId)) {
+ resourceConsumption = subResourceConsumptions.get(applicationId)
+ } else {
+ resourceConsumption = ResourceConsumption(0, 0, 0, 0)
+ }
}
resourceConsumption
}
