[incubator-celeborn] 01/01: [CELEBORN-302] Fix workers count out of sync in HA mode.

[et...@apache.org]

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

ethanfeng pushed a commit to branch CELEBORN-302
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git

commit 539fa6019a827f62a6eadb253258cc24cc5396e7
Author: Ethan Feng <et...@apache.org>
AuthorDate: Wed Feb 15 15:40:57 2023 +0800

    [CELEBORN-302] Fix workers count out of sync in HA mode.
---
 .../scala/org/apache/celeborn/common/util/PbSerDeUtils.scala     | 2 +-
 .../service/deploy/master/clustermeta/AbstractMetaManager.java   | 9 +++------
 2 files changed, 4 insertions(+), 7 deletions(-)

diff --git a/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala b/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
index e35725ab..fa033c60 100644
--- a/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
@@ -363,7 +363,7 @@ object PbSerDeUtils {
       blacklist: java.util.Set[WorkerInfo],
       workerLostEvent: java.util.Set[WorkerInfo],
       appHeartbeatTime: java.util.Map[String, java.lang.Long],
-      workers: java.util.ArrayList[WorkerInfo],
+      workers: java.util.Set[WorkerInfo],
       partitionTotalWritten: java.lang.Long,
       partitionTotalFileCount: java.lang.Long,
       appDiskUsageMetricSnapshots: Array[AppDiskUsageSnapShot],
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..585cf543 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
@@ -21,11 +21,7 @@ import static org.apache.celeborn.common.protocol.RpcNameConstants.WORKER_EP;
 
 import java.io.*;
 import java.nio.file.Files;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
+import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
@@ -51,7 +47,7 @@ public abstract class AbstractMetaManager implements IMetadataHandler {
   // Meta data for master service
   public final Set<String> registeredShuffle = ConcurrentHashMap.newKeySet();
   public final Set<String> hostnameSet = ConcurrentHashMap.newKeySet();
-  public final ArrayList<WorkerInfo> workers = new ArrayList<>();
+  public final Set<WorkerInfo> workers = new HashSet<>();
   public final ConcurrentHashMap<String, Long> appHeartbeatTime = new ConcurrentHashMap<>();
   // blacklist
   public final Set<WorkerInfo> blacklist = ConcurrentHashMap.newKeySet();
@@ -225,6 +221,7 @@ public abstract class AbstractMetaManager implements IMetadataHandler {
     workerInfo.lastHeartbeat_$eq(System.currentTimeMillis());
 
     try {
+      // if some master failed, other master not, sync state will be broken.
       workerInfo.setupEndpoint(rpcEnv.setupEndpointRef(RpcAddress.apply(host, rpcPort), WORKER_EP));
     } catch (Exception e) {
       LOG.error("Worker register failed", e);