HBASE-28158 update crashed time, unit test and checkstyle fix

Author: ukumawat

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

@@ -75,6 +75,10 @@ synchronized void putIfAbsent(ServerName sn) {
     this.deadServers.putIfAbsent(sn, EnvironmentEdgeManager.currentTime());
   }
 
+  synchronized void putIfAbsent(ServerName sn, Long crashedTime) {
+    this.deadServers.putIfAbsent(sn, crashedTime);
+  }
+
   public synchronized int size() {
     return deadServers.size();
   }

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

@@ -1033,7 +1033,8 @@ private void finishActiveMasterInitialization() throws IOException, InterruptedE
     Map<Class<?>, List<Procedure<MasterProcedureEnv>>> procsByType = procedureExecutor
       .getActiveProceduresNoCopy().stream().collect(Collectors.groupingBy(p -> p.getClass()));
 
-    // This manager must be accessed AFTER hbase:meta is confirmed on line..
+    // This manager must be accessed AFTER hbase:meta is confirmed on line.. and must be initialised
+    // before assignment manager
     this.tableStateManager = new TableStateManager(this);
     // Create Assignment Manager
     this.assignmentManager = createAssignmentManager(this, masterRegion);
@@ -1055,7 +1056,8 @@ private void finishActiveMasterInitialization() throws IOException, InterruptedE
     // TODO: Generate the splitting and live Set in one pass instead of two as we currently do.
     this.regionServerTracker.upgrade(
       procsByType.getOrDefault(ServerCrashProcedure.class, Collections.emptyList()).stream()
-        .map(p -> (ServerCrashProcedure) p).map(p -> p.getServerName()).collect(Collectors.toSet()),
+        .map(p -> (ServerCrashProcedure) p).collect(
+          Collectors.toMap(ServerCrashProcedure::getServerName, Procedure::getSubmittedTime)),
       Sets.union(rsListStorage.getAll(), walManager.getLiveServersFromWALDir()),
       walManager.getSplittingServersFromWALDir());
 

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

@@ -24,6 +24,7 @@
 import java.io.InterruptedIOException;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -116,22 +117,23 @@ private RegionServerInfo getServerInfo(ServerName serverName)
    * In this method, we will also construct the region server sets in {@link ServerManager}. If a
    * region server is dead between the crash of the previous master instance and the start of the
    * current master instance, we will schedule a SCP for it. This is done in
-   * {@link ServerManager#findDeadServersAndProcess(Set, Set)}, we call it here under the lock
+   * {@link ServerManager#findDeadServersAndProcess(Map, Set)}, we call it here under the lock
    * protection to prevent concurrency issues with server expiration operation.
    * @param deadServersFromPE          the region servers which already have SCP associated.
    * @param liveServersBeforeRestart   the live region servers we recorded before master restarts.
    * @param splittingServersFromWALDir Servers whose WALs are being actively 'split'.
    */
-  public void upgrade(Set<ServerName> deadServersFromPE, Set<ServerName> liveServersBeforeRestart,
-    Set<ServerName> splittingServersFromWALDir) throws KeeperException, IOException {
+  public void upgrade(Map<ServerName, Long> deadServersFromPE,
+    Set<ServerName> liveServersBeforeRestart, Set<ServerName> splittingServersFromWALDir)
+    throws KeeperException, IOException {
     LOG.info(
       "Upgrading RegionServerTracker to active master mode; {} have existing"
         + "ServerCrashProcedures, {} possibly 'live' servers, and {} 'splitting'.",
       deadServersFromPE.size(), liveServersBeforeRestart.size(), splittingServersFromWALDir.size());
     // deadServersFromPE is made from a list of outstanding ServerCrashProcedures.
     // splittingServersFromWALDir are being actively split -- the directory in the FS ends in
     // '-SPLITTING'. Each splitting server should have a corresponding SCP. Log if not.
-    splittingServersFromWALDir.stream().filter(s -> !deadServersFromPE.contains(s))
+    splittingServersFromWALDir.stream().filter(s -> !deadServersFromPE.containsKey(s))
       .forEach(s -> LOG.error("{} has no matching ServerCrashProcedure", s));
     // create ServerNode for all possible live servers from wal directory and master local region
     liveServersBeforeRestart

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

@@ -440,7 +440,7 @@ boolean checkAndRecordNewServer(final ServerName serverName, final ServerMetrics
    * @param deadServersFromPE     the region servers which already have a SCP associated.
    * @param liveServersFromWALDir the live region servers from wal directory.
    */
-  void findDeadServersAndProcess(Set<ServerName> deadServersFromPE,
+  void findDeadServersAndProcess(Map<ServerName, Long> deadServersFromPE,
     Set<ServerName> liveServersFromWALDir) {
     deadServersFromPE.forEach(deadservers::putIfAbsent);
     liveServersFromWALDir.stream().filter(sn -> !onlineServers.containsKey(sn))

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

@@ -23,6 +23,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -1882,7 +1883,16 @@ public void visitRegionState(Result result, final RegionInfo regionInfo, final S
       }
       // add regions to RIT while visiting the meta
       regionInTransitionTracker.handleRegionStateNodeOperation(regionNode);
-      if (master.getServerManager().isServerDead(regionNode.getRegionLocation())) {
+      // If region location of region belongs to a dead server mark the region crashed
+      if (
+        regionNode.getRegionLocation() != null
+          && master.getServerManager().isServerDead(regionNode.getRegionLocation())
+      ) {
+        Date timeOfCrash =
+          master.getServerManager().getDeadServers().getTimeOfDeath(regionNode.getRegionLocation());
+        if (timeOfCrash != null) {
+          regionNode.crashed(timeOfCrash.getTime());
+        }
         regionInTransitionTracker.regionCrashed(regionNode);
       }
     }
@@ -2317,10 +2327,12 @@ public CompletableFuture<Void> regionClosedAbnormally(RegionStateNode regionNode
   // As soon as a server a crashed, region hosting on that are un-available, this method helps to
   // track those un-available regions. This method can only be called from ServerCrashProcedure.
   public void markRegionsAsCrashed(List<RegionInfo> regionsOnCrashedServer,
-    ServerName crashedServerName) {
+    ServerCrashProcedure scp) {
+    ServerName crashedServerName = scp.getServerName();
     for (RegionInfo regionInfo : regionsOnCrashedServer) {
       RegionStateNode node = regionStates.getOrCreateRegionStateNode(regionInfo);
       if (node.getRegionLocation() == crashedServerName) {
+        node.crashed(scp.getSubmittedTime());
         regionInTransitionTracker.regionCrashed(node);
       }
     }

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

@@ -96,7 +96,7 @@ public AssignmentProcedureEvent(final RegionInfo regionInfo) {
 
   /**
    * Updated whenever a call to {@link #setRegionLocation(ServerName)} or
-   * {@link #setState(RegionState.State, RegionState.State...)}.
+   * {@link #setState(RegionState.State, RegionState.State...)} or {@link #crashed(long)}.
    */
   private volatile long lastUpdate = 0;
 
@@ -190,6 +190,10 @@ public void setLastHost(final ServerName serverName) {
     this.lastHost = serverName;
   }
 
+  public void crashed(long crashTime) {
+    this.lastUpdate = crashTime;
+  }
+
   public void setOpenSeqNum(final long seqId) {
     this.openSeqNum = seqId;
   }

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

@@ -208,8 +208,7 @@ protected Flow executeFromState(MasterProcedureEnv env, ServerCrashState state)
             if (LOG.isTraceEnabled()) {
               this.regionsOnCrashedServer.stream().forEach(ri -> LOG.trace(ri.getShortNameToLog()));
             }
-            env.getAssignmentManager().markRegionsAsCrashed(regionsOnCrashedServer,
-              this.serverName);
+            env.getAssignmentManager().markRegionsAsCrashed(regionsOnCrashedServer, this);
           }
           if (!this.shouldSplitWal) {
             setNextState(ServerCrashState.SERVER_CRASH_ASSIGN);

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

@@ -263,7 +263,9 @@ public void testBalancerWithoutFavoredNodes() throws Exception {
 
     // Balancer should unassign the region
     assertTrue("Balancer did not run", admin.balance());
-    TEST_UTIL.waitUntilNoRegionsInTransition();
+    TEST_UTIL.waitUntilNoRegionTransitScheduled();
+    assertEquals("One region should be unassigned", 1,
+      master.getAssignmentManager().getRegionsInTransitionCount());
 
     admin.assign(region.getEncodedNameAsBytes());
     TEST_UTIL.waitUntilNoRegionsInTransition(60000);

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestSplitTransactionOnCluster.java

@@ -1146,10 +1146,9 @@ public ReportRegionStateTransitionResponse reportRegionStateTransition(RpcContro
       ) {
         AssignmentManager am = myMaster.getAssignmentManager();
         for (RegionStateNode regionState : am.getRegionsInTransition()) {
-          /*
-           * TODO!!!! // Find the merging_new region and remove it if (regionState.isSplittingNew())
-           * { regionStates.deleteRegion(regionState.getRegion()); }
-           */
+          if (regionState.toRegionState().isSplittingNew()) {
+            am.getRegionStates().deleteRegion(regionState.toRegionState().getRegion());
+          }
         }
       }
       return resp;