From 6a3d220e9bd1edfd2e6a6eec77cab245560b861e Mon Sep 17 00:00:00 2001 From: stack Date: Tue, 19 Nov 2019 15:42:32 -0800 Subject: [PATCH] HBASE-23322 [hbck2] Simplification on HBCKSCP scheduling --- .../hbase/master/MasterRpcServices.java | 35 +++++------------ .../hadoop/hbase/master/ServerManager.java | 39 ++++--------------- .../master/assignment/AssignmentManager.java | 35 +++++++++++------ .../master/TestClusterRestartFailover.java | 6 ++- 4 files changed, 45 insertions(+), 70 deletions(-) diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterRpcServices.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterRpcServices.java index ac79cdb262..545dd868f8 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterRpcServices.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterRpcServices.java @@ -2555,35 +2555,18 @@ public class MasterRpcServices extends RSRpcServices public MasterProtos.ScheduleServerCrashProcedureResponse scheduleServerCrashProcedure( RpcController controller, MasterProtos.ScheduleServerCrashProcedureRequest request) throws ServiceException { - List serverNames = request.getServerNameList(); List pids = new ArrayList<>(); - try { - for (HBaseProtos.ServerName sn: serverNames) { - ServerName serverName = ProtobufUtil.toServerName(sn); - LOG.info("{} schedule ServerCrashProcedure for {}", - this.master.getClientIdAuditPrefix(), serverName); - if (shouldSubmitSCP(serverName)) { - final boolean containsMetaWALs = containMetaWals(serverName); - long pid = this.master.getServerManager().expireServer(serverName, - new Function() { - @Override - public Long apply(ServerName serverName) { - ProcedureExecutor procExec = - master.getMasterProcedureExecutor(); - return procExec.submitProcedure( - new HBCKServerCrashProcedure(procExec.getEnvironment(), - serverName, true, containsMetaWALs)); - } - }); - pids.add(pid); - } else { - pids.add(Procedure.NO_PROC_ID); - } + for (HBaseProtos.ServerName sn: request.getServerNameList()) { + ServerName serverName = ProtobufUtil.toServerName(sn); + LOG.info("{} schedule ServerCrashProcedure for {}", + this.master.getClientIdAuditPrefix(), serverName); + if (shouldSubmitSCP(serverName)) { + pids.add(this.master.getServerManager().expireServer(serverName)); + } else { + pids.add(Procedure.NO_PROC_ID); } - return MasterProtos.ScheduleServerCrashProcedureResponse.newBuilder().addAllPid(pids).build(); - } catch (IOException e) { - throw new ServiceException(e); } + return MasterProtos.ScheduleServerCrashProcedureResponse.newBuilder().addAllPid(pids).build(); } @Override diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java index 9c5ae64fca..8ecf3c4665 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java @@ -33,7 +33,6 @@ import java.util.concurrent.ConcurrentNavigableMap; import java.util.concurrent.ConcurrentSkipListMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.function.Function; import java.util.function.Predicate; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.ClockOutOfSyncException; @@ -532,32 +531,13 @@ public class ServerManager { /** * Expire the passed server. Add it to list of dead servers and queue a shutdown processing. - * @return True if we queued a ServerCrashProcedure else false if we did not (could happen for - * many reasons including the fact that its this server that is going down or we already - * have queued an SCP for this server or SCP processing is currently disabled because we - * are in startup phase). - */ - public boolean expireServer(final ServerName serverName) { - return expireServer(serverName, new Function() { - @Override - public Long apply(ServerName serverName) { - return master.getAssignmentManager().submitServerCrash(serverName, true); - } - }) != Procedure.NO_PROC_ID; - } - - /** - * Expire the passed server. Add it to list of dead servers and queue a shutdown processing. - * Used when expireServer is externally invoked by hbck2. - * @param function Takes ServerName and returns pid. See default implementation which queues - * an SCP via the AssignmentManager. - * @return True if we queued a ServerCrashProcedure else false if we did not (could happen for - * many reasons including the fact that its this server that is going down or we already - * have queued an SCP for this server or SCP processing is currently disabled because we - * are in startup phase). - */ - synchronized long expireServer(final ServerName serverName, - Function function) { + * @return pid if we queued a ServerCrashProcedure else {@link Procedure#NO_PROC_ID} if we did + * not (could happen for many reasons including the fact that its this server that is + * going down or we already have queued an SCP for this server or SCP processing is + * currently disabled because we are in startup phase). + */ + @VisibleForTesting // Redo test so we can make this protected. + public synchronized long expireServer(final ServerName serverName) { // THIS server is going down... can't handle our own expiration. if (serverName.equals(master.getServerName())) { if (!(master.isAborted() || master.isStopped())) { @@ -582,10 +562,7 @@ public class ServerManager { return Procedure.NO_PROC_ID; } LOG.info("Processing expiration of " + serverName + " on " + this.master.getServerName()); - long pid = function.apply(serverName); - if (pid <= 0) { - return Procedure.NO_PROC_ID; - } + long pid = master.getAssignmentManager().submitServerCrash(serverName, true); // Tell our listeners that a server was removed if (!this.listeners.isEmpty()) { this.listeners.stream().forEach(l -> l.serverRemoved(serverName)); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignmentManager.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignmentManager.java index 315b67b978..4421cf0ee4 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignmentManager.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignmentManager.java @@ -58,6 +58,7 @@ import org.apache.hadoop.hbase.master.RegionState.State; import org.apache.hadoop.hbase.master.ServerManager; import org.apache.hadoop.hbase.master.TableStateManager; import org.apache.hadoop.hbase.master.balancer.FavoredStochasticBalancer; +import org.apache.hadoop.hbase.master.procedure.HBCKServerCrashProcedure; import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv; import org.apache.hadoop.hbase.master.procedure.MasterProcedureScheduler; import org.apache.hadoop.hbase.master.procedure.ProcedureSyncWait; @@ -1485,14 +1486,14 @@ public class AssignmentManager { } public long submitServerCrash(ServerName serverName, boolean shouldSplitWal) { - boolean carryingMeta; - long pid; + // If serverNode is not known, its an 'Unknown Server' case. ServerStateNode serverNode = regionStates.getServerNode(serverName); if (serverNode == null) { - LOG.info("Skip to add SCP for {} since this server should be OFFLINE already", serverName); - return -1; + LOG.warn("Unknown {}", serverName); } + boolean carryingMeta; + long pid; // Remove the in-memory rsReports result synchronized (rsReports) { rsReports.remove(serverName); @@ -1502,26 +1503,38 @@ public class AssignmentManager { // server state to CRASHED, we will no longer accept the reportRegionStateTransition call from // this server. This is used to simplify the implementation for TRSP and SCP, where we can make // sure that, the region list fetched by SCP will not be changed any more. - serverNode.writeLock().lock(); + if (serverNode != null) { + serverNode.writeLock().lock(); + } try { ProcedureExecutor procExec = this.master.getMasterProcedureExecutor(); carryingMeta = isCarryingMeta(serverName); - if (!serverNode.isInState(ServerState.ONLINE)) { + if (serverNode != null && !serverNode.isInState(ServerState.ONLINE)) { LOG.info( "Skip to add SCP for {} with meta= {}, " + "since there should be a SCP is processing or already done for this server node", serverName, carryingMeta); return -1; } else { - serverNode.setState(ServerState.CRASHED); - pid = procExec.submitProcedure(new ServerCrashProcedure(procExec.getEnvironment(), - serverName, shouldSplitWal, carryingMeta)); + if (serverNode != null) { + serverNode.setState(ServerState.CRASHED); + } + // If serverNode == null, then 'Unknown Server'. Schedule HBCKSCP instead. + // It scours Master in-memory state AND hbase;meta for references to + // serverName just-in-case. An SCP that is scheduled when the server is + // 'Unknown' probably originated externally with HBCK2 fix-it tool. + MasterProcedureEnv mpe = procExec.getEnvironment(); + pid = procExec.submitProcedure(serverNode != null? + new ServerCrashProcedure(mpe, serverName, shouldSplitWal, carryingMeta): + new HBCKServerCrashProcedure(mpe, serverName, shouldSplitWal, carryingMeta)); LOG.info( - "Added {} to dead servers which carryingMeta={}, submitted ServerCrashProcedure pid={}", + "Added {} to dead servers, carryingMeta={}, submitted ServerCrashProcedure pid={}", serverName, carryingMeta, pid); } } finally { - serverNode.writeLock().unlock(); + if (serverNode != null) { + serverNode.writeLock().unlock(); + } } return pid; } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestClusterRestartFailover.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestClusterRestartFailover.java index 1577af84b5..2e18c16028 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestClusterRestartFailover.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestClusterRestartFailover.java @@ -109,7 +109,8 @@ public class TestClusterRestartFailover extends AbstractTestRestartCluster { ((ServerCrashProcedure) p).getServerName().equals(SERVER_FOR_TEST)).findAny(); assertTrue("Should have one SCP for " + SERVER_FOR_TEST, procedure.isPresent()); assertFalse("Submit the SCP for the same serverName " + SERVER_FOR_TEST + " which should fail", - UTIL.getHBaseCluster().getMaster().getServerManager().expireServer(SERVER_FOR_TEST)); + UTIL.getHBaseCluster().getMaster().getServerManager().expireServer(SERVER_FOR_TEST) == + Procedure.NO_PROC_ID); // Wait the SCP to finish SCP_LATCH.countDown(); @@ -117,7 +118,8 @@ public class TestClusterRestartFailover extends AbstractTestRestartCluster { assertFalse("Even when the SCP is finished, the duplicate SCP should not be scheduled for " + SERVER_FOR_TEST, - UTIL.getHBaseCluster().getMaster().getServerManager().expireServer(SERVER_FOR_TEST)); + UTIL.getHBaseCluster().getMaster().getServerManager().expireServer(SERVER_FOR_TEST) == + Procedure.NO_PROC_ID); serverNode = UTIL.getHBaseCluster().getMaster().getAssignmentManager().getRegionStates() .getServerNode(SERVER_FOR_TEST); assertNull("serverNode should be deleted after SCP finished", serverNode); -- 2.19.1