[server] Servers with PERMANENT_OFFLINE and TEMPORARY_OFFLINE tag are no longer assigned replicas

Author: Liebing

fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/CoordinatorEventProcessor.java

@@ -247,7 +247,9 @@ public void startup() {
         HashSet<ServerInfo> tabletServerInfoList =
                 new HashSet<>(coordinatorContext.getLiveTabletServers().values());
         serverMetadataCache.updateMetadata(
-                coordinatorContext.getCoordinatorServerInfo(), tabletServerInfoList);
+                coordinatorContext.getCoordinatorServerInfo(),
+                tabletServerInfoList,
+                coordinatorContext.getServerTags());
         updateTabletServerMetadataCacheWhenStartup(tabletServerInfoList);
 
         // start table manager
@@ -818,7 +820,8 @@ private void processNewTabletServer(NewTabletServerEvent newTabletServerEvent) {
         // update coordinatorServer metadata cache for the new added table server.
         serverMetadataCache.updateMetadata(
                 coordinatorContext.getCoordinatorServerInfo(),
-                new HashSet<>(coordinatorContext.getLiveTabletServers().values()));
+                new HashSet<>(coordinatorContext.getLiveTabletServers().values()),
+                coordinatorContext.getServerTags());
         // update server info for all tablet servers.
         updateTabletServerMetadataCache(
                 new HashSet<>(coordinatorContext.getLiveTabletServers().values()),
@@ -874,7 +877,9 @@ private void processDeadTabletServer(DeadTabletServerEvent deadTabletServerEvent
                 new HashSet<>(coordinatorContext.getLiveTabletServers().values());
         // update coordinatorServer metadata cache.
         serverMetadataCache.updateMetadata(
-                coordinatorContext.getCoordinatorServerInfo(), serverInfos);
+                coordinatorContext.getCoordinatorServerInfo(),
+                serverInfos,
+                coordinatorContext.getServerTags());
         updateTabletServerMetadataCache(serverInfos, null, null, Collections.emptySet());
 
         TableBucketStateMachine tableBucketStateMachine = tableManager.getTableBucketStateMachine();
@@ -947,6 +952,11 @@ private AddServerTagResponse processAddServerTag(AddServerTagEvent event) {
 
         // Then update coordinatorContext.
         serverIds.forEach(serverId -> coordinatorContext.putServerTag(serverId, serverTag));
+        // update coordinatorServer metadata cache for the new added serverTag
+        serverMetadataCache.updateMetadata(
+                coordinatorContext.getCoordinatorServerInfo(),
+                new HashSet<>(coordinatorContext.getLiveTabletServers().values()),
+                coordinatorContext.getServerTags());
 
         return addServerTagResponse;
     }
@@ -989,6 +999,11 @@ private RemoveServerTagResponse processRemoveServerTag(RemoveServerTagEvent even
 
         // Then update coordinatorContext.
         serverIds.forEach(coordinatorContext::removeServerTag);
+        // update coordinatorServer metadata cache for the new removed serverTag
+        serverMetadataCache.updateMetadata(
+                coordinatorContext.getCoordinatorServerInfo(),
+                new HashSet<>(coordinatorContext.getLiveTabletServers().values()),
+                coordinatorContext.getServerTags());
 
         return removeServerTagResponse;
     }

fluss-server/src/main/java/com/alibaba/fluss/server/metadata/CoordinatorMetadataCache.java

@@ -19,6 +19,7 @@
 
 import com.alibaba.fluss.cluster.ServerNode;
 import com.alibaba.fluss.cluster.TabletServerInfo;
+import com.alibaba.fluss.cluster.maintencance.ServerTag;
 import com.alibaba.fluss.server.coordinator.CoordinatorServer;
 
 import javax.annotation.Nullable;
@@ -27,11 +28,13 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
 
 import static com.alibaba.fluss.utils.concurrent.LockUtils.inLock;
 
@@ -42,7 +45,7 @@ public class CoordinatorMetadataCache implements ServerMetadataCache {
 
     @GuardedBy("metadataLock")
     private volatile NodeMetadataSnapshot metadataSnapshot =
-            new NodeMetadataSnapshot(null, Collections.emptyMap());
+            new NodeMetadataSnapshot(null, Collections.emptyMap(), Collections.emptyMap());
 
     public CoordinatorMetadataCache() {}
 
@@ -91,7 +94,33 @@ public Set<TabletServerInfo> getAliveTabletServerInfos() {
         return Collections.unmodifiableSet(tabletServerInfos);
     }
 
-    public void updateMetadata(ServerInfo coordinatorServer, Set<ServerInfo> serverInfoSet) {
+    @Override
+    public TabletServerInfo[] getLiveServers() {
+        Set<TabletServerInfo> aliveTabletServerInfosWithoutOfflineServerTag =
+                getAliveTabletServerInfos().stream()
+                        .filter(
+                                info ->
+                                        !metadataSnapshot.serverTags.containsKey(info.getId())
+                                                || (metadataSnapshot.serverTags.get(info.getId())
+                                                                != ServerTag.TEMPORARY_OFFLINE
+                                                        && metadataSnapshot.serverTags.get(
+                                                                        info.getId())
+                                                                != ServerTag.PERMANENT_OFFLINE))
+                        .collect(Collectors.toSet());
+        TabletServerInfo[] server =
+                new TabletServerInfo[aliveTabletServerInfosWithoutOfflineServerTag.size()];
+        Iterator<TabletServerInfo> iterator =
+                aliveTabletServerInfosWithoutOfflineServerTag.iterator();
+        for (int i = 0; i < aliveTabletServerInfosWithoutOfflineServerTag.size(); i++) {
+            server[i] = iterator.next();
+        }
+        return server;
+    }
+
+    public void updateMetadata(
+            ServerInfo coordinatorServer,
+            Set<ServerInfo> serverInfoSet,
+            Map<Integer, ServerTag> serverTagMap) {
         inLock(
                 metadataLock,
                 () -> {
@@ -101,19 +130,23 @@ public void updateMetadata(ServerInfo coordinatorServer, Set<ServerInfo> serverI
                     }
 
                     this.metadataSnapshot =
-                            new NodeMetadataSnapshot(coordinatorServer, newAliveTableServers);
+                            new NodeMetadataSnapshot(
+                                    coordinatorServer, newAliveTableServers, serverTagMap);
                 });
     }
 
     private static class NodeMetadataSnapshot {
         final @Nullable ServerInfo coordinatorServer;
         final Map<Integer, ServerInfo> aliveTabletServers;
+        final Map<Integer, ServerTag> serverTags;
 
         private NodeMetadataSnapshot(
                 @Nullable ServerInfo coordinatorServer,
-                Map<Integer, ServerInfo> aliveTabletServers) {
+                Map<Integer, ServerInfo> aliveTabletServers,
+                Map<Integer, ServerTag> serverTags) {
             this.coordinatorServer = coordinatorServer;
             this.aliveTabletServers = aliveTabletServers;
+            this.serverTags = serverTags;
         }
     }
 }

fluss-server/src/test/java/com/alibaba/fluss/server/metadata/CoordinatorMetadataCacheTest.java

@@ -20,12 +20,15 @@
 import com.alibaba.fluss.cluster.Endpoint;
 import com.alibaba.fluss.cluster.ServerType;
 import com.alibaba.fluss.cluster.TabletServerInfo;
+import com.alibaba.fluss.cluster.maintencance.ServerTag;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Set;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -36,6 +39,7 @@ public class CoordinatorMetadataCacheTest {
 
     private ServerInfo coordinatorServer;
     private Set<ServerInfo> aliveTableServers;
+    private Map<Integer, ServerTag> serverTagMap;
 
     @BeforeEach
     public void setup() {
@@ -68,11 +72,15 @@ public void setup() {
                                         "rack2",
                                         Endpoint.fromListenersString("INTERNAL://localhost:104"),
                                         ServerType.TABLET_SERVER)));
+
+        serverTagMap = new HashMap<>();
+        serverTagMap.put(0, ServerTag.PERMANENT_OFFLINE);
+        serverTagMap.put(1, ServerTag.TEMPORARY_OFFLINE);
     }
 
     @Test
     void testCoordinatorServerMetadataCache() {
-        serverMetadataCache.updateMetadata(coordinatorServer, aliveTableServers);
+        serverMetadataCache.updateMetadata(coordinatorServer, aliveTableServers, serverTagMap);
         assertThat(serverMetadataCache.getCoordinatorServer("CLIENT"))
                 .isEqualTo(coordinatorServer.node("CLIENT"));
         assertThat(serverMetadataCache.getCoordinatorServer("INTERNAL"))
@@ -85,5 +93,7 @@ void testCoordinatorServerMetadataCache() {
                         new TabletServerInfo(0, "rack0"),
                         new TabletServerInfo(1, "rack1"),
                         new TabletServerInfo(2, "rack2"));
+        assertThat(serverMetadataCache.getLiveServers())
+                .containsExactlyInAnyOrder(new TabletServerInfo(2, "rack2"));
     }
 }