You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by et...@apache.org on 2025/02/06 09:25:19 UTC
(celeborn) branch branch-0.5 updated: [CELEBORN-1850] Setup worker endpoint after initalizing controller
This is an automated email from the ASF dual-hosted git repository.
ethanfeng pushed a commit to branch branch-0.5
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/branch-0.5 by this push:
new 5287f5d0c [CELEBORN-1850] Setup worker endpoint after initalizing controller
5287f5d0c is described below
commit 5287f5d0c868f077e26a64e5cfff47ae68913772
Author: Sanskar Modi <sa...@gmail.com>
AuthorDate: Thu Feb 6 16:08:26 2025 +0800
[CELEBORN-1850] Setup worker endpoint after initalizing controller
### What changes were proposed in this pull request?
Setup worker endpoint after initalizing the controller.
### Why are the changes needed?
In current flow controller rpc endpoint is setup before calling the `controller.init(this)` leaving the required members to be unintialized which leading to lot of NullPointer exception in worker during restarts.
```
10:53:55.262 [celeborn-dispatcher-6] ERROR org.apache.celeborn.common.rpc.netty.Inbox - Ignoring error
java.lang.NullPointerException: null
at org.apache.celeborn.service.deploy.worker.Controller.org$apache$celeborn$service$deploy$worker$Controller$$handleDestroy(Controller.scala:659) ~[celeborn-worker_2.12-0.5.3.jar:0.5.3]
at org.apache.celeborn.service.deploy.worker.Controller$$anonfun$receiveAndReply$1.applyOrElse(Controller.scala:143) ~[celeborn-worker_2.12-0.5.3.jar:0.5.3]
at org.apache.celeborn.common.rpc.netty.Inbox.processInternal(Inbox.scala:119) ~[celeborn-common_2.12-0.5.3.jar:0.5.3]
at org.apache.celeborn.common.rpc.netty.Inbox.$anonfun$process$1(Inbox.scala:218) ~[celeborn-common_2.12-0.5.3.jar:0.5.3]
at org.apache.celeborn.common.rpc.netty.Inbox.safelyCall(Inbox.scala:314) ~[celeborn-common_2.12-0.5.3.jar:0.5.3]
at org.apache.celeborn.common.rpc.netty.Inbox.process(Inbox.scala:218) ~[celeborn-common_2.12-0.5.3.jar:0.5.3]
at org.apache.celeborn.common.rpc.netty.Dispatcher$MessageLoop.run(Dispatcher.scala:238) ~[celeborn-common_2.12-0.5.3.jar:0.5.3]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_412]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_412]
at java.lang.Thread.run(Thread.java:750) ~[?:1.8.0_412]
10:53:55.570 [main] INFO org.apache.celeborn.service.deploy.worker.Worker - Register worker successfully.
10:53:55.573 [main] INFO org.apache.celeborn.service.deploy.worker.Worker - Worker started.
```
### Does this PR introduce _any_ user-facing change?
NA
### How was this patch tested?
Existing UTs
Closes #3086 from s0nskar/null_controller.
Authored-by: Sanskar Modi <sa...@gmail.com>
Signed-off-by: mingji <fe...@alibaba-inc.com>
(cherry picked from commit fdf1883f2520f4e7a18f7bd934dece7f2cb5b2b3)
---
.../scala/org/apache/celeborn/service/deploy/worker/Worker.scala | 7 ++++---
1 file changed, 4 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 251422610..813ec60fe 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
@@ -194,8 +194,7 @@ private[celeborn] class Worker(
conf.workerCongestionControlCheckIntervalMs)
}
- var controller = new Controller(rpcEnv, conf, metricsSystem, workerSource)
- rpcEnv.setupEndpoint(RpcNameConstants.WORKER_EP, controller)
+ val controller = new Controller(rpcEnv, conf, metricsSystem, workerSource)
// Visible for testing
private[worker] var internalRpcEndpoint: RpcEndpoint = _
@@ -543,9 +542,11 @@ private[celeborn] class Worker(
pushDataHandler.init(this)
replicateHandler.init(this)
fetchHandler.init(this)
- controller.init(this)
workerStatusManager.init(this)
+ controller.init(this)
+ rpcEnv.setupEndpoint(RpcNameConstants.WORKER_EP, controller)
+
logInfo("Worker started.")
rpcEnv.awaitTermination()
if (conf.internalPortEnabled) {