This is an automated email from the ASF dual-hosted git repository. ethanfeng pushed a commit to branch branch-0.2 in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
commit d4208f9d5c5636cbb922e9e9ec43c3fabf4bcd17 Author: Ethan Feng <[email protected]> AuthorDate: Mon Feb 20 21:46:33 2023 +0800 [CELEBORN-302] Fix workers count out of sync in HA mode. (#1239) (cherry picked from commit 7e9ba19d583ebc8e89abee72a2622154ffc4152a) --- .../apache/celeborn/common/meta/WorkerInfo.scala | 18 ++++++------ .../master/clustermeta/AbstractMetaManager.java | 13 +++++++-- .../celeborn/service/deploy/master/Master.scala | 32 ++++++++++++---------- 3 files changed, 38 insertions(+), 25 deletions(-) diff --git a/common/src/main/scala/org/apache/celeborn/common/meta/WorkerInfo.scala b/common/src/main/scala/org/apache/celeborn/common/meta/WorkerInfo.scala index dca13b13..eabe9ae2 100644 --- a/common/src/main/scala/org/apache/celeborn/common/meta/WorkerInfo.scala +++ b/common/src/main/scala/org/apache/celeborn/common/meta/WorkerInfo.scala @@ -259,17 +259,19 @@ class WorkerInfo( |""".stripMargin } - override def equals(obj: Any): Boolean = { - val other = obj.asInstanceOf[WorkerInfo] - host == other.host && - rpcPort == other.rpcPort && - pushPort == other.pushPort && - fetchPort == other.fetchPort && - replicatePort == other.replicatePort + override def equals(other: Any): Boolean = other match { + case that: WorkerInfo => + host == that.host && + rpcPort == that.rpcPort && + pushPort == that.pushPort && + fetchPort == that.fetchPort && + replicatePort == that.replicatePort + case _ => false } override def hashCode(): Int = { - Objects.hashCode(host, rpcPort, pushPort, fetchPort, replicatePort) + val state = Seq(host, rpcPort, pushPort, fetchPort, replicatePort) + state.map(_.hashCode()).foldLeft(0)((a, b) => 31 * a + b) } } diff --git a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java index 75f18c26..19c3ce67 100644 --- a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java +++ b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java @@ -227,13 +227,20 @@ public abstract class AbstractMetaManager implements IMetadataHandler { try { workerInfo.setupEndpoint(rpcEnv.setupEndpointRef(RpcAddress.apply(host, rpcPort), WORKER_EP)); } catch (Exception e) { - LOG.error("Worker register failed", e); - return; + LOG.warn("Worker register setupEndpoint failed {}, will retry", e); + try { + workerInfo.setupEndpoint( + rpcEnv.setupEndpointRef(RpcAddress.apply(host, rpcPort), WORKER_EP)); + } catch (Exception e1) { + workerInfo.setupEndpoint(null); + } } workerInfo.updateDiskMaxSlots(estimatedPartitionSize); synchronized (workers) { - workers.add(workerInfo); + if (!workers.contains(workerInfo)) { + workers.add(workerInfo); + } } } diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala index 9952d8b7..4be15bcf 100644 --- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala +++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala @@ -706,32 +706,36 @@ private[celeborn] class Master( private def requestGetWorkerInfos(endpoint: RpcEndpointRef): GetWorkerInfosResponse = { try { - endpoint.askSync[GetWorkerInfosResponse](GetWorkerInfos) + if (endpoint != null) { + return endpoint.askSync[GetWorkerInfosResponse](GetWorkerInfos) + } } catch { case e: Exception => logError(s"AskSync GetWorkerInfos failed.", e) - val result = new util.ArrayList[WorkerInfo] - result.add(new WorkerInfo( - "unknown", - -1, - -1, - -1, - -1, - new util.HashMap[String, DiskInfo](), - new ConcurrentHashMap[UserIdentifier, ResourceConsumption](), - null)) - GetWorkerInfosResponse(StatusCode.REQUEST_FAILED, result.asScala: _*) } + val result = new util.ArrayList[WorkerInfo] + result.add(new WorkerInfo( + "unknown", + -1, + -1, + -1, + -1, + new util.HashMap[String, DiskInfo](), + new ConcurrentHashMap[UserIdentifier, ResourceConsumption](), + null)) + GetWorkerInfosResponse(StatusCode.REQUEST_FAILED, result.asScala: _*) } private def requestThreadDump(endpoint: RpcEndpointRef): ThreadDumpResponse = { try { - endpoint.askSync[ThreadDumpResponse](ThreadDump) + if (endpoint != null) { + return endpoint.askSync[ThreadDumpResponse](ThreadDump) + } } catch { case e: Exception => logError(s"AskSync ThreadDump failed.", e) - ThreadDumpResponse("Unknown") } + ThreadDumpResponse("Unknown") } private def isMasterActive: Int = {
