HBASE-25282 Remove processingServers in DeadServer as we can get this… (#2657)

Signed-off-by: Guanghao Zhang <[email protected]>
Change-Id: I804f1fa313a6ec8c0dc87fecc78046443c35a22a

Author: yuqi1129

hbase-server/src/main/java/org/apache/hadoop/hbase/master/DeadServer.java

@@ -33,8 +33,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
-
 /**
  * Class to hold dead servers list and utility querying dead server list. Servers are added when
  * they expire or when we find them in filesystem on startup. When a server crash procedure is
@@ -56,12 +54,6 @@ public class DeadServer {
    */
   private final Map<ServerName, Long> deadServers = new HashMap<>();
 
-  /**
-   * Set of dead servers currently being processed by a SCP. Added to this list at the start of SCP
-   * and removed after it is done processing the crash.
-   */
-  private final Set<ServerName> processingServers = new HashSet<>();
-
   /**
    * @param serverName server name.
    * @return true if this server is on the dead servers list false otherwise
@@ -70,15 +62,6 @@ public synchronized boolean isDeadServer(final ServerName serverName) {
     return deadServers.containsKey(serverName);
   }
 
-  /**
-   * Checks if there are currently any dead servers being processed by the master. Returns true if
-   * at least one region server is currently being processed as dead.
-   * @return true if any RS are being processed as dead
-   */
-  synchronized boolean areDeadServersInProgress() {
-    return !processingServers.isEmpty();
-  }
-
   public synchronized Set<ServerName> copyServerNames() {
     Set<ServerName> clone = new HashSet<>(deadServers.size());
     clone.addAll(deadServers.keySet());
@@ -90,29 +73,6 @@ public synchronized Set<ServerName> copyServerNames() {
    */
   synchronized void putIfAbsent(ServerName sn) {
     this.deadServers.putIfAbsent(sn, EnvironmentEdgeManager.currentTime());
-    processing(sn);
-  }
-
-  /**
-   * Add <code>sn<</code> to set of processing deadservers.
-   * @see #finish(ServerName)
-   */
-  public synchronized void processing(ServerName sn) {
-    if (processingServers.add(sn)) {
-      // Only log on add.
-      LOG.debug("Processing {}; numProcessing={}", sn, processingServers.size());
-    }
-  }
-
-  /**
-   * Complete processing for this dead server.
-   * @param sn ServerName for the dead server.
-   * @see #processing(ServerName)
-   */
-  public synchronized void finish(ServerName sn) {
-    if (processingServers.remove(sn)) {
-      LOG.debug("Removed {} from processing; numProcessing={}", sn, processingServers.size());
-    }
   }
 
   public synchronized int size() {
@@ -171,17 +131,12 @@ public synchronized String toString() {
     // Display unified set of servers from both maps
     Set<ServerName> servers = new HashSet<>();
     servers.addAll(deadServers.keySet());
-    servers.addAll(processingServers);
     StringBuilder sb = new StringBuilder();
     for (ServerName sn : servers) {
       if (sb.length() > 0) {
         sb.append(", ");
       }
       sb.append(sn.toString());
-      // Star entries that are being processed
-      if (processingServers.contains(sn)) {
-        sb.append("*");
-      }
     }
     return sb.toString();
   }
@@ -220,9 +175,6 @@ public synchronized Date getTimeOfDeath(final ServerName deadServerName) {
    * @return true if this server was removed
    */
   public synchronized boolean removeDeadServer(final ServerName deadServerName) {
-    Preconditions.checkState(!processingServers.contains(deadServerName),
-      "Asked to remove server still in processingServers set " + deadServerName + " (numProcessing="
-        + processingServers.size() + ")");
     return this.deadServers.remove(deadServerName) != null;
   }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterRpcServices.java

@@ -2528,10 +2528,17 @@ public ClearDeadServersResponse clearDeadServers(RpcController controller,
         response.addAllServerName(request.getServerNameList());
       } else {
         for (HBaseProtos.ServerName pbServer : request.getServerNameList()) {
-          if (
-            !master.getServerManager().getDeadServers()
-              .removeDeadServer(ProtobufUtil.toServerName(pbServer))
-          ) {
+          ServerName server = ProtobufUtil.toServerName(pbServer);
+
+          final boolean deadInProcess = master.getProcedures().stream().anyMatch(
+            p -> (p instanceof ServerCrashProcedure)
+              && ((ServerCrashProcedure) p).getServerName().equals(server));
+          if (deadInProcess) {
+            throw new ServiceException(
+              String.format("Dead server '%s' is not 'dead' in fact...", server));
+          }
+
+          if (!deadServer.removeDeadServer(server)) {
             response.addServerName(pbServer);
           }
         }

hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java

@@ -50,6 +50,8 @@
 import org.apache.hadoop.hbase.ipc.HBaseRpcController;
 import org.apache.hadoop.hbase.ipc.RemoteWithExtrasException;
 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
+import org.apache.hadoop.hbase.master.assignment.RegionStates;
+import org.apache.hadoop.hbase.master.procedure.ServerCrashProcedure;
 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
 import org.apache.hadoop.hbase.procedure2.Procedure;
 import org.apache.hadoop.hbase.regionserver.HRegionServer;
@@ -547,8 +549,8 @@ public DeadServer getDeadServers() {
    * Checks if any dead servers are currently in progress.
    * @return true if any RS are being processed as dead, false if not
    */
-  public boolean areDeadServersInProgress() {
-    return this.deadservers.areDeadServersInProgress();
+  public boolean areDeadServersInProgress() throws IOException {
+    return master.getProcedures().stream().anyMatch(p -> p instanceof ServerCrashProcedure);
   }
 
   void letRegionServersShutdown() {

hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/ServerCrashProcedure.java

@@ -140,7 +140,6 @@ protected Flow executeFromState(MasterProcedureEnv env, ServerCrashState state)
     // This adds server to the DeadServer processing list but not to the DeadServers list.
     // Server gets removed from processing list below on procedure successful finish.
     if (!notifiedDeadServer) {
-      services.getServerManager().getDeadServers().processing(serverName);
       notifiedDeadServer = true;
     }
 
@@ -255,7 +254,6 @@ protected Flow executeFromState(MasterProcedureEnv env, ServerCrashState state)
         case SERVER_CRASH_FINISH:
           LOG.info("removed crashed server {} after splitting done", serverName);
           services.getAssignmentManager().getRegionStates().removeServer(serverName);
-          services.getServerManager().getDeadServers().finish(serverName);
           updateProgress(true);
           return Flow.NO_MORE_STATE;
         default:

hbase-server/src/test/java/org/apache/hadoop/hbase/TestRegionRebalancing.java

@@ -183,7 +183,7 @@ public void testRebalanceOnRegionServerNumberChange() throws IOException, Interr
   /**
    * Wait on crash processing. Balancer won't run if processing a crashed server.
    */
-  private void waitOnCrashProcessing() {
+  private void waitOnCrashProcessing() throws IOException {
     while (UTIL.getHBaseCluster().getMaster().getServerManager().areDeadServersInProgress()) {
       LOG.info("Waiting on processing of crashed server before proceeding...");
       Threads.sleep(1000);

hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestDeadServer.java

@@ -20,6 +20,7 @@
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
+import java.io.IOException;
 import java.util.List;
 import java.util.Set;
 import org.apache.hadoop.hbase.HBaseClassTestRule;
@@ -69,22 +70,10 @@ public static void tearDownAfterClass() throws Exception {
   public void testIsDead() {
     DeadServer ds = new DeadServer();
     ds.putIfAbsent(hostname123);
-    ds.processing(hostname123);
-    assertTrue(ds.areDeadServersInProgress());
-    ds.finish(hostname123);
-    assertFalse(ds.areDeadServersInProgress());
 
     ds.putIfAbsent(hostname1234);
-    ds.processing(hostname1234);
-    assertTrue(ds.areDeadServersInProgress());
-    ds.finish(hostname1234);
-    assertFalse(ds.areDeadServersInProgress());
 
     ds.putIfAbsent(hostname12345);
-    ds.processing(hostname12345);
-    assertTrue(ds.areDeadServersInProgress());
-    ds.finish(hostname12345);
-    assertFalse(ds.areDeadServersInProgress());
 
     // Already dead = 127.0.0.1,9090,112321
     // Coming back alive = 127.0.0.1,9090,223341
@@ -104,15 +93,15 @@ public void testIsDead() {
   }
 
   @Test
-  public void testCrashProcedureReplay() {
+  public void testCrashProcedureReplay() throws IOException {
     HMaster master = TEST_UTIL.getHBaseCluster().getMaster();
     final ProcedureExecutor<MasterProcedureEnv> pExecutor = master.getMasterProcedureExecutor();
     ServerCrashProcedure proc =
       new ServerCrashProcedure(pExecutor.getEnvironment(), hostname123, false, false);
 
     ProcedureTestingUtility.submitAndWait(pExecutor, proc);
 
-    assertFalse(master.getServerManager().getDeadServers().areDeadServersInProgress());
+    assertTrue(master.getServerManager().areDeadServersInProgress());
   }
 
   @Test
@@ -163,17 +152,14 @@ public void testClearDeadServer() {
     d.putIfAbsent(hostname1234);
     Assert.assertEquals(2, d.size());
 
-    d.finish(hostname123);
     d.removeDeadServer(hostname123);
     Assert.assertEquals(1, d.size());
-    d.finish(hostname1234);
     d.removeDeadServer(hostname1234);
     Assert.assertTrue(d.isEmpty());
 
     d.putIfAbsent(hostname1234);
     Assert.assertFalse(d.removeDeadServer(hostname123_2));
     Assert.assertEquals(1, d.size());
-    d.finish(hostname1234);
     Assert.assertTrue(d.removeDeadServer(hostname1234));
     Assert.assertTrue(d.isEmpty());
   }

hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestRollingRestart.java

@@ -230,6 +230,8 @@ private boolean isDeadServerSCPExecuted(ServerName serverName) throws IOExceptio
 
   private void waitForRSShutdownToStartAndFinish(MasterThread activeMaster, ServerName serverName)
     throws InterruptedException, IOException {
+  private void waitForRSShutdownToStartAndFinish(MasterThread activeMaster,
+      ServerName serverName) throws InterruptedException, IOException {
     ServerManager sm = activeMaster.getMaster().getServerManager();
     // First wait for it to be in dead list
     while (!sm.getDeadServers().isDeadServer(serverName)) {

hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestHBCKSCP.java

@@ -145,7 +145,6 @@ public void test() throws Exception {
     cluster.killRegionServer(rsServerName);
 
     master.getServerManager().moveFromOnlineToDeadServers(rsServerName);
-    master.getServerManager().getDeadServers().finish(rsServerName);
     master.getServerManager().getDeadServers().removeDeadServer(rsServerName);
     master.getAssignmentManager().getRegionStates().removeServer(rsServerName);
     // Kill the server. Nothing should happen since an 'Unknown Server' as far

hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestSyncReplicationStandbyKillRS.java

@@ -0,0 +1,120 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.replication;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.master.MasterFileSystem;
+import org.apache.hadoop.hbase.master.ServerManager;
+import org.apache.hadoop.hbase.testclassification.LargeTests;
+import org.apache.hadoop.hbase.testclassification.ReplicationTests;
+import org.apache.hadoop.hbase.util.JVMClusterUtil;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Category({ ReplicationTests.class, LargeTests.class })
+public class TestSyncReplicationStandbyKillRS extends SyncReplicationTestBase {
+
+  private static final Logger LOG =
+      LoggerFactory.getLogger(TestSyncReplicationStandbyKillRS.class);
+
+  private final long SLEEP_TIME = 1000;
+
+  private final int COUNT = 1000;
+
+  @ClassRule
+  public static final HBaseClassTestRule CLASS_RULE =
+      HBaseClassTestRule.forClass(TestSyncReplicationStandbyKillRS.class);
+
+  @Test
+  public void testStandbyKillRegionServer() throws Exception {
+    MasterFileSystem mfs = UTIL2.getHBaseCluster().getMaster().getMasterFileSystem();
+    Path remoteWALDir = getRemoteWALDir(mfs, PEER_ID);
+    assertFalse(mfs.getWALFileSystem().exists(remoteWALDir));
+    UTIL2.getAdmin().transitReplicationPeerSyncReplicationState(PEER_ID,
+        SyncReplicationState.STANDBY);
+    assertTrue(mfs.getWALFileSystem().exists(remoteWALDir));
+    UTIL1.getAdmin().transitReplicationPeerSyncReplicationState(PEER_ID,
+        SyncReplicationState.ACTIVE);
+
+    // Disable async replication and write data, then shutdown
+    UTIL1.getAdmin().disableReplicationPeer(PEER_ID);
+    write(UTIL1, 0, COUNT);
+    UTIL1.shutdownMiniCluster();
+
+    JVMClusterUtil.MasterThread activeMaster = UTIL2.getMiniHBaseCluster().getMasterThread();
+    Thread t = new Thread(() -> {
+      try {
+        List<JVMClusterUtil.RegionServerThread> regionServers =
+            UTIL2.getMiniHBaseCluster().getLiveRegionServerThreads();
+        for (JVMClusterUtil.RegionServerThread rst : regionServers) {
+          ServerName serverName = rst.getRegionServer().getServerName();
+          rst.getRegionServer().stop("Stop RS for test");
+          waitForRSShutdownToStartAndFinish(activeMaster, serverName);
+          JVMClusterUtil.RegionServerThread restarted =
+              UTIL2.getMiniHBaseCluster().startRegionServer();
+          restarted.waitForServerOnline();
+        }
+      } catch (Exception e) {
+        LOG.error("Failed to kill RS", e);
+      }
+    });
+    t.start();
+
+    // Transit standby to DA to replay logs
+    try {
+      UTIL2.getAdmin().transitReplicationPeerSyncReplicationState(PEER_ID,
+        SyncReplicationState.DOWNGRADE_ACTIVE);
+    } catch (Exception e) {
+      LOG.error("Failed to transit standby cluster to " + SyncReplicationState.DOWNGRADE_ACTIVE, e);
+    }
+
+    while (UTIL2.getAdmin().getReplicationPeerSyncReplicationState(PEER_ID)
+        != SyncReplicationState.DOWNGRADE_ACTIVE) {
+      Thread.sleep(SLEEP_TIME);
+    }
+    verify(UTIL2, 0, COUNT);
+  }
+
+  private void waitForRSShutdownToStartAndFinish(JVMClusterUtil.MasterThread activeMaster,
+      ServerName serverName) throws InterruptedException, IOException {
+    ServerManager sm = activeMaster.getMaster().getServerManager();
+    // First wait for it to be in dead list
+    while (!sm.getDeadServers().isDeadServer(serverName)) {
+      LOG.debug("Waiting for [" + serverName + "] to be listed as dead in master");
+      Thread.sleep(SLEEP_TIME);
+    }
+    LOG.debug("Server [" + serverName + "] marked as dead, waiting for it to " +
+        "finish dead processing");
+    while (sm.areDeadServersInProgress()) {
+      LOG.debug("Server [" + serverName + "] still being processed, waiting");
+      Thread.sleep(SLEEP_TIME);
+    }
+    LOG.debug("Server [" + serverName + "] done with server shutdown processing");
+  }
+}