[incubator-celeborn] branch main updated: [CELEBORN-302] Fix workers count out of sync in HA mode. (#1239)

[zhouky]

This is an automated email from the ASF dual-hosted git repository.

zhouky 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 7e9ba19d [CELEBORN-302] Fix workers count out of sync in HA mode. 
(#1239)
7e9ba19d is described below

commit 7e9ba19d583ebc8e89abee72a2622154ffc4152a
Author: Ethan Feng <ethanf...@apache.org>
AuthorDate: Mon Feb 20 21:46:33 2023 +0800

    [CELEBORN-302] Fix workers count out of sync in HA mode. (#1239)
---
 .../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 0e1a7453..79603a31 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
@@ -751,32 +751,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 = {