retry policy

Author: dengzhhu653

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java

@@ -9167,11 +9167,11 @@ public Map<String, String> updateTableColumnStatistics(ColumnStatistics colStats
       openTransaction();
       // DataNucleus objects get detached all over the place for no (real) reason.
       // So let's not use them anywhere unless absolutely necessary.
-      int maxRetries = MetastoreConf.getIntVar(conf, ConfVars.METASTORE_S4U_NOWAIT_MAX_RETRIES);
-      long sleepInterval = ThreadLocalRandom.current().nextLong(MetastoreConf.getTimeVar(conf,
-          ConfVars.METASTORE_S4U_NOWAIT_RETRY_SLEEP_INTERVAL, TimeUnit.MILLISECONDS)) + 30;
       MTable mTable = ensureGetMTable(catName, statsDesc.getDbName(), statsDesc.getTableName());
-      Map<String, String> result = new RetryingExecutor<>(maxRetries, sleepInterval, () -> {
+      int maxRetries = MetastoreConf.getIntVar(conf, ConfVars.METASTORE_S4U_NOWAIT_MAX_RETRIES);
+      long sleepInterval = MetastoreConf.getTimeVar(conf,
+          ConfVars.METASTORE_S4U_NOWAIT_RETRY_SLEEP_INTERVAL, TimeUnit.MILLISECONDS);
+      Map<String, String> result = new RetryingExecutor<>(maxRetries, () -> {
         Ref<Exception> exceptionRef = new Ref<>();
         String savePoint = "uts_" + ThreadLocalRandom.current().nextInt(10000) + "_" + System.nanoTime();
         setTransactionSavePoint(savePoint);
@@ -9232,7 +9232,9 @@ public Map<String, String> updateTableColumnStatistics(ColumnStatistics colStats
         }
         oldt.setParameters(newParams);
         return newParams;
-      }).onRetry(RetryingExecutor.RetryException.class).commandName("updateTableColumnStatistics").run();
+      }).onRetry(e -> e instanceof RetryingExecutor.RetryException)
+        .commandName("updateTableColumnStatistics").sleepInterval(sleepInterval, interval ->
+              ThreadLocalRandom.current().nextLong(sleepInterval) + 30).run();
       committed = commitTransaction();
       // TODO: similar to update...Part, this used to do "return committed;"; makes little sense.
       return committed ? result : null;
@@ -11111,17 +11113,17 @@ private void lockNotificationSequenceForUpdate() throws MetaException {
       // Derby doesn't allow FOR UPDATE to lock the row being selected (See https://db.apache
       // .org/derby/docs/10.1/ref/rrefsqlj31783.html) . So lock the whole table. Since there's
       // only one row in the table, this shouldn't cause any performance degradation.
-      new RetryingExecutor<Void>(maxRetries, sleepInterval, () -> {
+      new RetryingExecutor<Void>(maxRetries, () -> {
         directSql.lockDbTable("NOTIFICATION_SEQUENCE");
         return null;
-      }).commandName("lockNotificationSequenceForUpdate").run();
+      }).commandName("lockNotificationSequenceForUpdate").sleepInterval(sleepInterval).run();
     } else {
       String selectQuery = "select \"NEXT_EVENT_ID\" from \"NOTIFICATION_SEQUENCE\"";
       String lockingQuery = sqlGenerator.addForUpdateClause(selectQuery);
-      new RetryingExecutor<Void>(maxRetries, sleepInterval, () -> {
+      new RetryingExecutor<Void>(maxRetries, () -> {
         executePlainSQL(lockingQuery, null);
         return null;
-      }).commandName("lockNotificationSequenceForUpdate").run();
+      }).commandName("lockNotificationSequenceForUpdate").sleepInterval(sleepInterval).run();
     }
   }
 

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/utils/RetryingExecutor.java

@@ -18,32 +18,41 @@
 
 package org.apache.hadoop.hive.metastore.utils;
 
-import java.util.ArrayList;
-import java.util.List;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.UndeclaredThrowableException;
 import java.util.concurrent.Callable;
+import java.util.function.Function;
+import java.util.function.Predicate;
 
+import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.apache.hadoop.hive.metastore.api.MetaException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class RetryingExecutor<T> {
-  private static Logger LOG = LoggerFactory.getLogger(RetryingExecutor.class);
+  private static final Logger LOG = LoggerFactory.getLogger(RetryingExecutor.class);
 
   private final int maxRetries;
-  private final long sleepInterval;
+  private long sleepInterval = 1000;
   private final Callable<T> command;
-  private final List<Class<? extends Exception>> retriableException = new ArrayList<>();
+  private Predicate<Exception> retryPolicy;
   private int currentRetries = 0;
   private String commandName;
+  private Function<Long, Long> sleepIntervalFunc;
 
-  public RetryingExecutor(int maxRetries, long sleepInterval, Callable<T> command) {
+  public RetryingExecutor(int maxRetries, Callable<T> command) {
     this.maxRetries = maxRetries;
-    this.sleepInterval = sleepInterval;
     this.command = command;
+    // default commandName unless specified
+    this.commandName = StackWalker.getInstance()
+        .walk(frames -> frames
+            .skip(1)
+            .findFirst()
+            .map(StackWalker.StackFrame::getMethodName)).get();
   }
 
-  public RetryingExecutor<T> onRetry(Class<? extends Exception> ex) {
-    this.retriableException.add(ex);
+  public RetryingExecutor<T> onRetry(Predicate<Exception> retryPolicy) {
+    this.retryPolicy = retryPolicy;
     return this;
   }
 
@@ -52,6 +61,17 @@ public RetryingExecutor<T> commandName(String name) {
     return this;
   }
 
+  public RetryingExecutor<T> sleepInterval(long sleepInterval) {
+    return sleepInterval(sleepInterval, null);
+  }
+
+  public RetryingExecutor<T> sleepInterval(long sleepInterval,
+      Function<Long, Long> sleepIntervalFunc) {
+    this.sleepInterval = sleepInterval;
+    this.sleepIntervalFunc = sleepIntervalFunc;
+    return this;
+  }
+
   public T run() throws MetaException {
     while (true) {
       try {
@@ -68,7 +88,7 @@ public T run() throws MetaException {
         }
         currentRetries++;
         try {
-          Thread.sleep(sleepInterval);
+          Thread.sleep(getSleepInterval());
         } catch (InterruptedException e1) {
           String msg = "Couldn't run the command: " + commandName + " in " + currentRetries +
               " retry, because the following error: ";
@@ -80,11 +100,16 @@ public T run() throws MetaException {
   }
 
   private void checkException(Exception e) throws MetaException {
-    if (!retriableException.isEmpty() &&
-        retriableException.stream().noneMatch(nex -> nex.isInstance(e))) {
-      String message = "See a non-retriable exception, avoid to retry the command:" + commandName;
+    if (retryPolicy != null && !retryPolicy.test(e)) {
+      String message = "See a fatal exception, avoid to retry the command:" + commandName;
       LOG.info(message, e);
-      throw new MetaException(message + " :: " + e.getMessage());
+      String errorMessage = ExceptionUtils.getMessage(e);
+      if (e instanceof InvocationTargetException || e instanceof UndeclaredThrowableException) {
+        errorMessage = ExceptionUtils.getMessage(e.getCause());
+      }
+      Throwable rootCause = ExceptionUtils.getRootCause(e);
+      errorMessage += (rootCause == null ? "" : ("\nRoot cause: " + rootCause));
+      throw new MetaException(message + " :: " + errorMessage);
     }
   }
 
@@ -101,6 +126,9 @@ public RetryException(String msg) {
   }
 
   public long getSleepInterval() {
+    if (sleepIntervalFunc != null) {
+      this.sleepInterval = sleepIntervalFunc.apply(sleepInterval);
+    }
     return sleepInterval;
   }
-}
+}

standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestObjectStore.java

@@ -1277,7 +1277,8 @@ public void testRetryingExecutorSleep() throws Exception {
         MetastoreConf.getIntVar(conf, ConfVars.NOTIFICATION_SEQUENCE_LOCK_MAX_RETRIES);
     long sleepInterval = MetastoreConf.getTimeVar(conf,
         ConfVars.NOTIFICATION_SEQUENCE_LOCK_RETRY_SLEEP_INTERVAL, TimeUnit.MILLISECONDS);
-    RetryingExecutor<Void> re = new RetryingExecutor<>(maxRetries, sleepInterval, null);
+    RetryingExecutor<Void> re = new RetryingExecutor<Void>(maxRetries, null)
+        .sleepInterval(sleepInterval);
     Assert.assertTrue("invalid sleep value", re.getSleepInterval() >= 0);
   }