From f4c91ee65bd16459cfbd4b5dabdf3f953bea968b Mon Sep 17 00:00:00 2001
From: Xiaolin Ha <haxiaolin@apache.org>
Date: Mon, 7 Mar 2022 16:41:49 +0800
Subject: [PATCH] HBASE-26552 Introduce retry to logroller to avoid abort
 (#4170)

Signed-off-by: Andrew Purtell <apurtell@apache.org>
---
 .../hadoop/hbase/wal/AbstractWALRoller.java   | 46 +++++++++++++++++--
 1 file changed, 43 insertions(+), 3 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractWALRoller.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractWALRoller.java
index 53dfa85fe4aa..1f14b18363b5 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractWALRoller.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractWALRoller.java
@@ -59,6 +59,19 @@ public abstract class AbstractWALRoller<T extends Abortable> extends Thread
 
   protected static final String WAL_ROLL_PERIOD_KEY = "hbase.regionserver.logroll.period";
 
+  /**
+   * Configure for the timeout of log rolling retry.
+   */
+  protected static final String WAL_ROLL_WAIT_TIMEOUT =
+    "hbase.regionserver.logroll.wait.timeout.ms";
+
+  /**
+   * Configure for the max count of log rolling retry.
+   * The real retry count is also limited by the timeout of log rolling
+   * via {@link #WAL_ROLL_WAIT_TIMEOUT}
+   */
+  protected static final String WAL_ROLL_RETRIES = "hbase.regionserver.logroll.retries";
+
   protected final ConcurrentMap<WAL, RollController> wals = new ConcurrentHashMap<>();
   protected final T abortable;
   // Period to roll log.
@@ -66,6 +79,10 @@ public abstract class AbstractWALRoller<T extends Abortable> extends Thread
   private final int threadWakeFrequency;
   // The interval to check low replication on hlog's pipeline
   private final long checkLowReplicationInterval;
+  // Wait period for roll log
+  private final long rollWaitTimeout;
+  // Max retry for roll log
+  private final int maxRollRetry;
 
   private volatile boolean running = true;
 
@@ -113,6 +130,9 @@ protected AbstractWALRoller(String name, Configuration conf, T abortable) {
     this.threadWakeFrequency = conf.getInt(HConstants.THREAD_WAKE_FREQUENCY, 10 * 1000);
     this.checkLowReplicationInterval =
       conf.getLong("hbase.regionserver.hlog.check.lowreplication.interval", 30 * 1000);
+    this.rollWaitTimeout = conf.getLong(WAL_ROLL_WAIT_TIMEOUT, 30000);
+    // retry rolling does not have to be the default behavior, so the default value is 0 here
+    this.maxRollRetry = conf.getInt(WAL_ROLL_RETRIES, 0);
   }
 
   /**
@@ -183,9 +203,29 @@ public void run() {
           } else {
             continue;
           }
-          // Force the roll if the logroll.period is elapsed or if a roll was requested.
-          // The returned value is an collection of actual region and family names.
-          Map<byte[], List<byte[]>> regionsToFlush = controller.rollWal(now);
+          Map<byte[], List<byte[]>> regionsToFlush = null;
+          int nAttempts = 0;
+          long startWaiting = System.currentTimeMillis();
+          do {
+            try {
+              // Force the roll if the logroll.period is elapsed or if a roll was requested.
+              // The returned value is an collection of actual region and family names.
+              regionsToFlush = controller.rollWal(System.currentTimeMillis());
+              break;
+            } catch (IOException ioe) {
+              long waitingTime = System.currentTimeMillis() - startWaiting;
+              if (waitingTime < rollWaitTimeout && nAttempts < maxRollRetry) {
+                nAttempts++;
+                LOG.warn("Retry to roll log, nAttempts={}, waiting time={}ms, sleeping 1s to retry,"
+                    + " last excepiton= {}", nAttempts, waitingTime,
+                  ioe.getCause().getClass().getSimpleName());
+                sleep(1000);
+              } else {
+                LOG.error("Roll wal failed and waiting timeout, will not retry", ioe);
+                throw ioe;
+              }
+            }
+          } while (EnvironmentEdgeManager.currentTime() - startWaiting < rollWaitTimeout);
           if (regionsToFlush != null) {
             for (Map.Entry<byte[], List<byte[]>> r : regionsToFlush.entrySet()) {
               scheduleFlush(Bytes.toString(r.getKey()), r.getValue());