HBASE-22360 Abort timer doesn't set when abort is called during graceful shutdown process

Signed-off-by: Sergey Shelukhin <sershe@apache.org>

Author: Bahram Chehrazy

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java

@@ -537,6 +537,9 @@ public class HRegionServer extends HasThread implements
   /**regionserver codec list **/
   public static final String REGIONSERVER_CODEC = "hbase.regionserver.codecs";
 
+  // A timer to shutdown the process if abort takes too long
+  private Timer abortMonitor;
+
   /**
    * Starts a HRegionServer at the default location
    */
@@ -1047,21 +1050,6 @@ public void run() {
       }
     }
 
-    if (abortRequested) {
-      Timer abortMonitor = new Timer("Abort regionserver monitor", true);
-      TimerTask abortTimeoutTask = null;
-      try {
-        abortTimeoutTask =
-            Class.forName(conf.get(ABORT_TIMEOUT_TASK, SystemExitWhenAbortTimeout.class.getName()))
-                .asSubclass(TimerTask.class).getDeclaredConstructor().newInstance();
-      } catch (Exception e) {
-        LOG.warn("Initialize abort timeout task failed", e);
-      }
-      if (abortTimeoutTask != null) {
-        abortMonitor.schedule(abortTimeoutTask, conf.getLong(ABORT_TIMEOUT, DEFAULT_ABORT_TIMEOUT));
-      }
-    }
-
     if (this.leases != null) {
       this.leases.closeAfterLeasesExpire();
     }
@@ -2448,6 +2436,8 @@ public void abort(String reason, Throwable cause) {
     } catch (Throwable t) {
       LOG.warn("Unable to report fatal error to master", t);
     }
+
+    scheduleAbortTimer();
     // shutdown should be run as the internal user
     stop(reason, true, null);
   }
@@ -2485,6 +2475,24 @@ protected void kill() {
   protected void sendShutdownInterrupt() {
   }
 
+  // Limits the time spent in the shutdown process.
+  private void scheduleAbortTimer() {
+    if (this.abortMonitor == null) {
+      this.abortMonitor = new Timer("Abort regionserver monitor", true);
+      TimerTask abortTimeoutTask = null;
+      try {
+        abortTimeoutTask =
+          Class.forName(conf.get(ABORT_TIMEOUT_TASK, SystemExitWhenAbortTimeout.class.getName()))
+            .asSubclass(TimerTask.class).getDeclaredConstructor().newInstance();
+      } catch (Exception e) {
+        LOG.warn("Initialize abort timeout task failed", e);
+      }
+      if (abortTimeoutTask != null) {
+        abortMonitor.schedule(abortTimeoutTask, conf.getLong(ABORT_TIMEOUT, DEFAULT_ABORT_TIMEOUT));
+      }
+    }
+  }
+
   /**
    * Wait on all threads to finish. Presumption is that all closes and stops
    * have already been called.