Fixed tags updated on restarts

Author: s0nskar

common/src/main/proto/TransportMessages.proto

@@ -221,7 +221,6 @@ message PbMetaRegisterWorkerRequest {
   map<string, PbResourceConsumption> userResourceConsumption = 7;
   int32 internalPort = 8;
   string networkLocation = 9;
-  repeated string tags = 12;
 }
 
 message PbHeartbeatFromWorker {

master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java

@@ -381,7 +381,10 @@ public void updateRegisterWorkerMeta(
     }
     workerInfo.updateDiskSlots(estimatedPartitionSize);
     synchronized (workersMap) {
-      workersMap.putIfAbsent(workerInfo.toUniqueId(), workerInfo);
+      WorkerInfo existingWorkerInfo = workersMap.putIfAbsent(workerInfo.toUniqueId(), workerInfo);
+      if (existingWorkerInfo != null) {
+        existingWorkerInfo.tags_$eq(new HashSet<>(tags));
+      }
       shutdownWorkers.remove(workerInfo);
       lostWorkers.remove(workerInfo);
       excludedWorkers.remove(workerInfo);