HBASE-19994 Create a new class for RPC throttling exception, make it retryable

hbase-client/src/main/java/org/apache/hadoop/hbase/exceptions/ClientExceptionsUtil.java

@@ -39,7 +39,7 @@ import org.apache.yetus.audience.InterfaceAudience;
 import org.apache.yetus.audience.InterfaceStability;
 import org.apache.hadoop.hbase.ipc.CallTimeoutException;
 import org.apache.hadoop.hbase.ipc.FailedServerException;
-import org.apache.hadoop.hbase.quotas.ThrottlingException;
+import org.apache.hadoop.hbase.quotas.RpcThrottlingException;
 import org.apache.hadoop.ipc.RemoteException;
 
 @InterfaceAudience.Private
@@ -60,7 +60,7 @@ public final class ClientExceptionsUtil {
 
   public static boolean isSpecialException(Throwable cur) {
     return (cur instanceof RegionMovedException || cur instanceof RegionOpeningException
-        || cur instanceof RegionTooBusyException || cur instanceof ThrottlingException
+        || cur instanceof RegionTooBusyException || cur instanceof RpcThrottlingException
         || cur instanceof MultiActionResultTooLarge || cur instanceof RetryImmediatelyException
         || cur instanceof CallQueueTooBigException || cur instanceof CallDroppedException
         || cur instanceof NotServingRegionException || cur instanceof RequestTooBigException);
@@ -73,7 +73,7 @@ public final class ClientExceptionsUtil {
    * - nested exceptions
    *
    * Looks for: RegionMovedException / RegionOpeningException / RegionTooBusyException /
-   *            ThrottlingException
+   *            RpcThrottlingException
    * @return null if we didn't find the exception, the exception otherwise.
    */
   public static Throwable findException(Object exception) {

hbase-client/src/main/java/org/apache/hadoop/hbase/quotas/RpcThrottlingException.java

@@ -0,0 +1,131 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable
+ * law or agreed to in writing, software distributed under the License is distributed on an "AS IS"
+ * BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License
+ * for the specific language governing permissions and limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.quotas;
+
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.apache.hadoop.hbase.HBaseIOException;
+import org.apache.hadoop.util.StringUtils;
+
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * Describe the throttling result. TODO: At some point this will be handled on the client side to
+ * prevent operation to go on the server if the waitInterval is greater than the one got as result
+ * of this exception.
+ */
+@InterfaceAudience.Public
+public class RpcThrottlingException extends HBaseIOException {
+
+  @InterfaceAudience.Public
+  public enum Type {
+    NumRequestsExceeded, RequestSizeExceeded, NumReadRequestsExceeded, NumWriteRequestsExceeded,
+    WriteSizeExceeded, ReadSizeExceeded,
+  }
+
+  private static final String[] MSG_TYPE =
+      new String[] { "number of requests exceeded", "request size limit exceeded",
+        "number of read requests exceeded", "number of write requests exceeded",
+        "write size limit exceeded", "read size limit exceeded", };
+
+  private static final String MSG_WAIT = " - wait ";
+
+  private long waitInterval;
+  private Type type;
+
+  public RpcThrottlingException(String msg) {
+    super(msg);
+
+    // Dirty workaround to get the information after
+    // ((RemoteException)e.getCause()).unwrapRemoteException()
+    for (int i = 0; i < MSG_TYPE.length; ++i) {
+      int index = msg.indexOf(MSG_TYPE[i]);
+      if (index >= 0) {
+        String waitTimeStr = msg.substring(index + MSG_TYPE[i].length() + MSG_WAIT.length());
+        type = Type.values()[i];
+        waitInterval = timeFromString(waitTimeStr);
+        break;
+      }
+    }
+  }
+
+  public RpcThrottlingException(final Type type, final long waitInterval, final String msg) {
+    super(msg);
+    this.waitInterval = waitInterval;
+    this.type = type;
+  }
+
+  public Type getType() {
+    return this.type;
+  }
+
+  public long getWaitInterval() {
+    return this.waitInterval;
+  }
+
+  public static void throwNumRequestsExceeded(final long waitInterval) throws
+      RpcThrottlingException {
+    throwThrottlingException(Type.NumRequestsExceeded, waitInterval);
+  }
+
+  public static void throwRequestSizeExceeded(final long waitInterval)
+      throws RpcThrottlingException {
+    throwThrottlingException(Type.RequestSizeExceeded, waitInterval);
+  }
+
+  public static void throwNumReadRequestsExceeded(final long waitInterval)
+      throws RpcThrottlingException {
+    throwThrottlingException(Type.NumReadRequestsExceeded, waitInterval);
+  }
+
+  public static void throwNumWriteRequestsExceeded(final long waitInterval)
+      throws RpcThrottlingException {
+    throwThrottlingException(Type.NumWriteRequestsExceeded, waitInterval);
+  }
+
+  public static void throwWriteSizeExceeded(final long waitInterval) throws RpcThrottlingException {
+    throwThrottlingException(Type.WriteSizeExceeded, waitInterval);
+  }
+
+  public static void throwReadSizeExceeded(final long waitInterval) throws RpcThrottlingException {
+    throwThrottlingException(Type.ReadSizeExceeded, waitInterval);
+  }
+
+  private static void throwThrottlingException(final Type type, final long waitInterval)
+      throws RpcThrottlingException {
+    String msg = MSG_TYPE[type.ordinal()] + MSG_WAIT + StringUtils.formatTime(waitInterval);
+    throw new RpcThrottlingException(type, waitInterval, msg);
+  }
+
+  private static long timeFromString(String timeDiff) {
+    Pattern[] patterns =
+        new Pattern[] { Pattern.compile("^(\\d+\\.\\d\\d)sec"),
+            Pattern.compile("^(\\d+)mins, (\\d+\\.\\d\\d)sec"),
+            Pattern.compile("^(\\d+)hrs, (\\d+)mins, (\\d+\\.\\d\\d)sec") };
+
+    for (int i = 0; i < patterns.length; ++i) {
+      Matcher m = patterns[i].matcher(timeDiff);
+      if (m.find()) {
+        long time = Math.round(Float.parseFloat(m.group(1 + i)) * 1000);
+        if (i > 0) {
+          time += Long.parseLong(m.group(i)) * (60 * 1000);
+        }
+        if (i > 1) {
+          time += Long.parseLong(m.group(i - 1)) * (60 * 60 * 1000);
+        }
+        return time;
+      }
+    }
+
+    return -1;
+  }
+}

hbase-client/src/main/java/org/apache/hadoop/hbase/quotas/ThrottlingException.java

@@ -29,7 +29,10 @@ import org.apache.yetus.audience.InterfaceAudience;
  * TODO: At some point this will be handled on the client side to prevent
  * operation to go on the server if the waitInterval is grater than the one got
  * as result of this exception.
+ *
+ * @deprecated  replaced by {@link RpcThrottlingException} since hbase-2.0.0.
  */
+@Deprecated
 @InterfaceAudience.Public
 public class ThrottlingException extends QuotaExceededException {
   private static final long serialVersionUID = 1406576492085155743L;

hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/DefaultOperationQuota.java

@@ -59,7 +59,7 @@ public class DefaultOperationQuota implements OperationQuota {
 
   @Override
   public void checkQuota(int numWrites, int numReads, int numScans)
-      throws ThrottlingException {
+      throws RpcThrottlingException {
     writeConsumed = estimateConsume(OperationType.MUTATE, numWrites, 100);
     readConsumed  = estimateConsume(OperationType.GET, numReads, 100);
     readConsumed += estimateConsume(OperationType.SCAN, numScans, 1000);

hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/NoopOperationQuota.java

@@ -43,7 +43,7 @@ class NoopOperationQuota implements OperationQuota {
 
   @Override
   public void checkQuota(int numWrites, int numReads, int numScans)
-      throws ThrottlingException {
+      throws RpcThrottlingException {
     // no-op
   }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/NoopQuotaLimiter.java

@@ -36,7 +36,7 @@ class NoopQuotaLimiter implements QuotaLimiter {
 
   @Override
   public void checkQuota(long estimateWriteSize, long estimateReadSize)
-      throws ThrottlingException {
+      throws RpcThrottlingException {
     // no-op
   }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/OperationQuota.java

@@ -41,10 +41,11 @@ public interface OperationQuota {
    * @param numWrites number of write operation that will be performed
    * @param numReads number of small-read operation that will be performed
    * @param numScans number of long-read operation that will be performed
-   * @throws ThrottlingException if the operation cannot be performed
+   * @throws RpcThrottlingException if the operation cannot be performed because
+   *   RPC quota is exceeded.
    */
   void checkQuota(int numWrites, int numReads, int numScans)
-    throws ThrottlingException;
+    throws RpcThrottlingException;
 
   /** Cleanup method on operation completion */
   void close();

hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/QuotaLimiter.java

@@ -33,10 +33,10 @@ public interface QuotaLimiter {
    *
    * @param estimateWriteSize the write size that will be checked against the available quota
    * @param estimateReadSize the read size that will be checked against the available quota
-   * @throws ThrottlingException thrown if not enough avialable resources to perform operation.
+   * @throws RpcThrottlingException thrown if not enough avialable resources to perform operation.
    */
   void checkQuota(long estimateWriteSize, long estimateReadSize)
-    throws ThrottlingException;
+    throws RpcThrottlingException;
 
   /**
    * Removes the specified write and read amount from the quota.

hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/RegionServerRpcQuotaManager.java

@@ -127,10 +127,10 @@ public class RegionServerRpcQuotaManager {
    * @param region the region where the operation will be performed
    * @param type the operation type
    * @return the OperationQuota
-   * @throws ThrottlingException if the operation cannot be executed due to quota exceeded.
+   * @throws RpcThrottlingException if the operation cannot be executed due to quota exceeded.
    */
   public OperationQuota checkQuota(final Region region,
-      final OperationQuota.OperationType type) throws IOException, ThrottlingException {
+      final OperationQuota.OperationType type) throws IOException, RpcThrottlingException {
     switch (type) {
       case SCAN:   return checkQuota(region, 0, 0, 1);
       case GET:    return checkQuota(region, 0, 1, 0);
@@ -146,10 +146,10 @@ public class RegionServerRpcQuotaManager {
    * @param region the region where the operation will be performed
    * @param actions the "multi" actions to perform
    * @return the OperationQuota
-   * @throws ThrottlingException if the operation cannot be executed due to quota exceeded.
+   * @throws RpcThrottlingException if the operation cannot be executed due to quota exceeded.
    */
   public OperationQuota checkQuota(final Region region,
-      final List<ClientProtos.Action> actions) throws IOException, ThrottlingException {
+      final List<ClientProtos.Action> actions) throws IOException, RpcThrottlingException {
     int numWrites = 0;
     int numReads = 0;
     for (final ClientProtos.Action action: actions) {
@@ -171,11 +171,11 @@ public class RegionServerRpcQuotaManager {
    * @param numReads number of short-reads to perform
    * @param numScans number of scan to perform
    * @return the OperationQuota
-   * @throws ThrottlingException if the operation cannot be executed due to quota exceeded.
+   * @throws RpcThrottlingException if the operation cannot be executed due to quota exceeded.
    */
   private OperationQuota checkQuota(final Region region,
       final int numWrites, final int numReads, final int numScans)
-      throws IOException, ThrottlingException {
+      throws IOException, RpcThrottlingException {
     Optional<User> user = RpcServer.getRequestUser();
     UserGroupInformation ugi;
     if (user.isPresent()) {
@@ -188,7 +188,7 @@ public class RegionServerRpcQuotaManager {
     OperationQuota quota = getQuota(ugi, table);
     try {
       quota.checkQuota(numWrites, numReads, numScans);
-    } catch (ThrottlingException e) {
+    } catch (RpcThrottlingException e) {
       LOG.debug("Throttling exception for user=" + ugi.getUserName() +
                 " table=" + table + " numWrites=" + numWrites +
                 " numReads=" + numReads + " numScans=" + numScans +

hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TimeBasedLimiter.java

@@ -110,30 +110,30 @@ public class TimeBasedLimiter implements QuotaLimiter {
   }
 
   @Override
-  public void checkQuota(long writeSize, long readSize) throws ThrottlingException {
+  public void checkQuota(long writeSize, long readSize) throws RpcThrottlingException {
     if (!reqsLimiter.canExecute()) {
-      ThrottlingException.throwNumRequestsExceeded(reqsLimiter.waitInterval());
+      RpcThrottlingException.throwNumRequestsExceeded(reqsLimiter.waitInterval());
     }
     if (!reqSizeLimiter.canExecute(writeSize + readSize)) {
-      ThrottlingException.throwRequestSizeExceeded(reqSizeLimiter
+      RpcThrottlingException.throwRequestSizeExceeded(reqSizeLimiter
           .waitInterval(writeSize + readSize));
     }
 
     if (writeSize > 0) {
       if (!writeReqsLimiter.canExecute()) {
-        ThrottlingException.throwNumWriteRequestsExceeded(writeReqsLimiter.waitInterval());
+        RpcThrottlingException.throwNumWriteRequestsExceeded(writeReqsLimiter.waitInterval());
       }
       if (!writeSizeLimiter.canExecute(writeSize)) {
-        ThrottlingException.throwWriteSizeExceeded(writeSizeLimiter.waitInterval(writeSize));
+        RpcThrottlingException.throwWriteSizeExceeded(writeSizeLimiter.waitInterval(writeSize));
       }
     }
 
     if (readSize > 0) {
       if (!readReqsLimiter.canExecute()) {
-        ThrottlingException.throwNumReadRequestsExceeded(readReqsLimiter.waitInterval());
+        RpcThrottlingException.throwNumReadRequestsExceeded(readReqsLimiter.waitInterval());
       }
       if (!readSizeLimiter.canExecute(readSize)) {
-        ThrottlingException.throwReadSizeExceeded(readSizeLimiter.waitInterval(readSize));
+        RpcThrottlingException.throwReadSizeExceeded(readSizeLimiter.waitInterval(readSize));
       }
     }
   }

hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestMetaCache.java

@@ -31,7 +31,7 @@ import org.apache.hadoop.hbase.*;
 import org.apache.hadoop.hbase.HBaseClassTestRule;
 import org.apache.hadoop.hbase.exceptions.ClientExceptionsUtil;
 import org.apache.hadoop.hbase.exceptions.RegionOpeningException;
-import org.apache.hadoop.hbase.quotas.ThrottlingException;
+import org.apache.hadoop.hbase.quotas.RpcThrottlingException;
 import org.apache.hadoop.hbase.regionserver.HRegionServer;
 import org.apache.hadoop.hbase.regionserver.RSRpcServices;
 import org.apache.hadoop.hbase.testclassification.ClientTests;
@@ -193,12 +193,12 @@ public class TestMetaCache {
 
   public static List<Throwable> metaCachePreservingExceptions() {
     return new ArrayList<Throwable>() {{
-      add(new RegionOpeningException(" "));
+        add(new RegionOpeningException(" "));
-      add(new RegionTooBusyException("Some old message"));
+        add(new RegionTooBusyException("Some old message"));
-      add(new ThrottlingException(" "));
+        add(new RpcThrottlingException(" "));
-      add(new MultiActionResultTooLarge(" "));
+        add(new MultiActionResultTooLarge(" "));
-      add(new RetryImmediatelyException(" "));
+        add(new RetryImmediatelyException(" "));
-      add(new CallQueueTooBigException());
+        add(new CallQueueTooBigException());
     }};
   }
 

hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaAdmin.java

@@ -22,8 +22,12 @@ import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
+
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellScanner;
 import org.apache.hadoop.hbase.HBaseClassTestRule;
@@ -31,13 +35,18 @@ import org.apache.hadoop.hbase.HBaseTestingUtility;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
+import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.ResultScanner;
 import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
 import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.testclassification.ClientTests;
 import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.util.Bytes;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -99,10 +108,10 @@ public class TestQuotaAdmin {
     Admin admin = TEST_UTIL.getAdmin();
     String userName = User.getCurrent().getShortName();
 
+    admin.setQuota(
+        QuotaSettingsFactory.throttleUser(userName, ThrottleType.READ_NUMBER, 6, TimeUnit.MINUTES));
     admin.setQuota(QuotaSettingsFactory
-      .throttleUser(userName, ThrottleType.READ_NUMBER, 6, TimeUnit.MINUTES));
+        .throttleUser(userName, ThrottleType.WRITE_NUMBER, 12, TimeUnit.MINUTES));
-    admin.setQuota(QuotaSettingsFactory
-      .throttleUser(userName, ThrottleType.WRITE_NUMBER, 12, TimeUnit.MINUTES));
     admin.setQuota(QuotaSettingsFactory.bypassGlobals(userName, true));
 
     try (QuotaRetriever scanner = QuotaRetriever.open(TEST_UTIL.getConfiguration())) {
@@ -183,6 +192,49 @@ public class TestQuotaAdmin {
     assertNumResults(0, null);
   }
 
+  @Test
+  public void testMultiQuotaThrottling() throws Exception {
+    byte[] FAMILY = Bytes.toBytes("testFamily");
+    byte[] ROW = Bytes.toBytes("testRow");
+    byte[] QUALIFIER = Bytes.toBytes("testQualifier");
+    byte[] VALUE = Bytes.toBytes("testValue");
+
+    Admin admin = TEST_UTIL.getAdmin();
+    TableName tableName = TableName.valueOf("testMultiQuotaThrottling");
+    TableDescriptor desc = TableDescriptorBuilder.newBuilder(tableName)
+        .setColumnFamily(ColumnFamilyDescriptorBuilder.of(FAMILY)).build();
+    admin.createTable(desc);
+
+    // Set up the quota.
+    admin.setQuota(QuotaSettingsFactory.throttleTable(tableName, ThrottleType.WRITE_NUMBER, 6,
+        TimeUnit.SECONDS));
+
+    Thread.sleep(1000);
+    TEST_UTIL.getRSForFirstRegionInTable(tableName).getRegionServerRpcQuotaManager().
+        getQuotaCache().triggerCacheRefresh();
+    Thread.sleep(1000);
+
+    Table t =  TEST_UTIL.getConnection().getTable(tableName);
+    try {
+      int size = 5;
+      List actions = new ArrayList();
+      Object[] results = new Object[size];
+
+      for (int i = 0; i < size; i++) {
+        Put put1 = new Put(ROW);
+        put1.addColumn(FAMILY, QUALIFIER, VALUE);
+        actions.add(put1);
+      }
+      t.batch(actions, results);
+      t.batch(actions, results);
+    } catch (IOException e) {
+      fail("Not supposed to get ThrottlingExcepiton " + e);
+    } finally {
+      t.close();
+    }
+  }
+
+
   @Test
   public void testQuotaRetrieverFilter() throws Exception {
     Admin admin = TEST_UTIL.getAdmin();
@@ -321,8 +373,8 @@ public class TestQuotaAdmin {
     final TableName tn = TableName.valueOf("sq_table2");
     final long originalSizeLimit = 1024L * 1024L * 1024L * 1024L * 5L; // 5TB
     final SpaceViolationPolicy violationPolicy = SpaceViolationPolicy.NO_WRITES;
-    QuotaSettings settings = QuotaSettingsFactory.limitTableSpace(
+    QuotaSettings settings = QuotaSettingsFactory.limitTableSpace(tn, originalSizeLimit,
-        tn, originalSizeLimit, violationPolicy);
+        violationPolicy);
     admin.setQuota(settings);
 
     // Verify the Quotas in the table
@@ -349,8 +401,8 @@ public class TestQuotaAdmin {
     // Setting a new size and policy should be reflected
     final long newSizeLimit = 1024L * 1024L * 1024L * 1024L; // 1TB
     final SpaceViolationPolicy newViolationPolicy = SpaceViolationPolicy.NO_WRITES_COMPACTIONS;
-    QuotaSettings newSettings = QuotaSettingsFactory.limitTableSpace(
+    QuotaSettings newSettings = QuotaSettingsFactory.limitTableSpace(tn, newSizeLimit,
-        tn, newSizeLimit, newViolationPolicy);
+        newViolationPolicy);
     admin.setQuota(newSettings);
 
     // Verify the new Quotas in the table

hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaState.java

@@ -216,7 +216,7 @@ public class TestQuotaState {
     try {
       limiter.checkQuota(1, 1);
       fail("Should have thrown ThrottlingException");
-    } catch (ThrottlingException e) {
+    } catch (RpcThrottlingException e) {
       // expected
     }
   }
@@ -225,7 +225,7 @@ public class TestQuotaState {
     for (int i = 0; i < availReqs; ++i) {
       try {
         limiter.checkQuota(1, 1);
-      } catch (ThrottlingException e) {
+      } catch (RpcThrottlingException e) {
         fail("Unexpected ThrottlingException after " + i + " requests. limit=" + availReqs);
       }
       limiter.grabQuota(1, 1);

hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaThrottle.java

@@ -520,7 +520,7 @@ public class TestQuotaThrottle {
         }
         count += tables.length;
       }
-    } catch (ThrottlingException e) {
+    } catch (RpcThrottlingException e) {
       LOG.error("put failed after nRetries=" + count, e);
     }
     return count;
@@ -536,7 +536,7 @@ public class TestQuotaThrottle {
         }
         count += tables.length;
       }
-    } catch (ThrottlingException e) {
+    } catch (RpcThrottlingException e) {
       LOG.error("get failed after nRetries=" + count, e);
     }
     return count;