Use BBS as default. Makes tests a bit more defensive against flakiness

Author: Ray Mattingly

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

@@ -51,14 +51,14 @@ public class DefaultOperationQuota implements OperationQuota {
   protected long readDiff = 0;
   protected long writeCapacityUnitDiff = 0;
   protected long readCapacityUnitDiff = 0;
-  private boolean useBlockBytesScanned;
+  private boolean useResultSizeBytes;
   private long blockSizeBytes;
 
   public DefaultOperationQuota(final Configuration conf, final int blockSizeBytes,
     final QuotaLimiter... limiters) {
     this(conf, Arrays.asList(limiters));
-    this.useBlockBytesScanned =
-      conf.getBoolean(OperationQuota.USE_BLOCK_BYTES_SCANNED_KEY, USE_BLOCK_BYTES_SCANNED_DEFAULT);
+    this.useResultSizeBytes =
+      conf.getBoolean(OperationQuota.USE_RESULT_SIZE_BYTES, USE_RESULT_SIZE_BYTES_DEFAULT);
     this.blockSizeBytes = blockSizeBytes;
   }
 
@@ -105,12 +105,12 @@ public void close() {
 
     long resultSize =
       operationSize[OperationType.GET.ordinal()] + operationSize[OperationType.SCAN.ordinal()];
-    if (useBlockBytesScanned) {
+    if (useResultSizeBytes) {
+      readDiff = resultSize - readConsumed;
+    } else {
       long blockBytesScanned =
         RpcServer.getCurrentCall().map(RpcCall::getBlockBytesScanned).orElse(0L);
       readDiff = Math.max(blockBytesScanned, resultSize) - readConsumed;
-    } else {
-      readDiff = resultSize - readConsumed;
     }
 
     writeCapacityUnitDiff =
@@ -158,13 +158,13 @@ public void addMutation(final Mutation mutation) {
   protected void updateEstimateConsumeQuota(int numWrites, int numReads, int numScans) {
     writeConsumed = estimateConsume(OperationType.MUTATE, numWrites, 100);
 
-    if (useBlockBytesScanned) {
+    if (useResultSizeBytes) {
+      readConsumed = estimateConsume(OperationType.GET, numReads, 100);
+      readConsumed += estimateConsume(OperationType.SCAN, numScans, 1000);
+    } else {
       // assume 1 block required for reads. this is probably a low estimate, which is okay
       readConsumed = numReads > 0 ? blockSizeBytes : 0;
       readConsumed += numScans > 0 ? blockSizeBytes : 0;
-    } else {
-      readConsumed = estimateConsume(OperationType.GET, numReads, 100);
-      readConsumed += estimateConsume(OperationType.SCAN, numScans, 1000);
     }
 
     writeCapacityUnitConsumed = calculateWriteCapacityUnit(writeConsumed);

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

@@ -35,8 +35,15 @@ public enum OperationType {
     SCAN
   }
 
-  String USE_BLOCK_BYTES_SCANNED_KEY = "hbase.quota.use.block.bytes.scanned";
-  boolean USE_BLOCK_BYTES_SCANNED_DEFAULT = false;
+  /**
+   * If false, the default, then IO based throttles will consume read availability based on the
+   * block bytes scanned by the given request. If true then IO based throttles will use result size
+   * rather than block bytes scanned. Using block bytes scanned should be preferable to using result
+   * size, because otherwise access patterns like heavily filtered scans may be able to produce a
+   * significant and effectively un-throttled workload.
+   */
+  String USE_RESULT_SIZE_BYTES = "hbase.quota.use.result.size.bytes";
+  boolean USE_RESULT_SIZE_BYTES_DEFAULT = false;
 
   /**
    * Checks if it is possible to execute the specified operation. The quota will be estimated based

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

@@ -23,9 +23,8 @@
 import static org.apache.hadoop.hbase.quotas.ThrottleQuotaTestUtil.doScans;
 import static org.apache.hadoop.hbase.quotas.ThrottleQuotaTestUtil.triggerUserCacheRefresh;
 import static org.apache.hadoop.hbase.quotas.ThrottleQuotaTestUtil.waitMinuteQuota;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
 
+import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
 import org.apache.hadoop.hbase.HBaseClassTestRule;
 import org.apache.hadoop.hbase.HBaseTestingUtil;
@@ -71,7 +70,6 @@ public static void setUpBeforeClass() throws Exception {
     // quotas enabled, using block bytes scanned
     TEST_UTIL.getConfiguration().setBoolean(QuotaUtil.QUOTA_CONF_KEY, true);
     TEST_UTIL.getConfiguration().setInt(QuotaCache.REFRESH_CONF_KEY, REFRESH_TIME);
-    TEST_UTIL.getConfiguration().setBoolean(OperationQuota.USE_BLOCK_BYTES_SCANNED_KEY, true);
 
     // don't cache blocks to make IO predictable
     TEST_UTIL.getConfiguration().setFloat(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, 0.0f);
@@ -106,22 +104,22 @@ public void testBBSGet() throws Exception {
     TEST_UTIL.flush(TABLE_NAME);
 
     // Add ~10 block/min limit
-    admin.setQuota(QuotaSettingsFactory.throttleUser(userName, ThrottleType.REQUEST_SIZE,
+    admin.setQuota(QuotaSettingsFactory.throttleUser(userName, ThrottleType.READ_SIZE,
       Math.round(10.1 * blockSize), TimeUnit.MINUTES));
     triggerUserCacheRefresh(TEST_UTIL, false, TABLE_NAME);
 
     // should execute at max 10 requests
-    assertEquals(10, doGets(20, FAMILY, QUALIFIER, table));
+    testTraffic(() -> doGets(20, FAMILY, QUALIFIER, table), 10, 1);
 
     // wait a minute and you should get another 10 requests executed
     waitMinuteQuota();
-    assertEquals(10, doGets(20, FAMILY, QUALIFIER, table));
+    testTraffic(() -> doGets(20, FAMILY, QUALIFIER, table), 10, 1);
 
     // Remove all the limits
     admin.setQuota(QuotaSettingsFactory.unthrottleUser(userName));
     triggerUserCacheRefresh(TEST_UTIL, true, TABLE_NAME);
-    assertEquals(100, doGets(100, FAMILY, QUALIFIER, table));
-    assertEquals(100, doGets(100, FAMILY, QUALIFIER, table));
+    testTraffic(() -> doGets(100, FAMILY, QUALIFIER, table), 100, 0);
+    testTraffic(() -> doGets(100, FAMILY, QUALIFIER, table), 100, 0);
   }
 
   @Test
@@ -142,30 +140,27 @@ public void testBBSScan() throws Exception {
     waitMinuteQuota();
 
     // should execute 1 request
-    assertEquals(1, doScans(5, table));
+    testTraffic(() -> doScans(5, table), 1, 0);
 
     // Remove all the limits
     admin.setQuota(QuotaSettingsFactory.unthrottleUser(userName));
     triggerUserCacheRefresh(TEST_UTIL, true, TABLE_NAME);
-    assertEquals(100, doScans(100, table));
-    assertEquals(100, doScans(100, table));
+    testTraffic(() -> doScans(100, table), 100, 0);
+    testTraffic(() -> doScans(100, table), 100, 0);
 
     // Add ~3 block/min limit. This should support >1 scans
     admin.setQuota(QuotaSettingsFactory.throttleUser(userName, ThrottleType.REQUEST_SIZE,
       Math.round(3.1 * blockSize), TimeUnit.MINUTES));
     triggerUserCacheRefresh(TEST_UTIL, false, TABLE_NAME);
 
     // should execute some requests, but not all
-    long successfulScans = doScans(100, table);
-    LOG.info("successfulScans = " + successfulScans);
-    assertTrue(successfulScans < 100);
-    assertTrue(successfulScans > 0);
+    testTraffic(() -> doScans(100, table), 100, 90);
 
     // Remove all the limits
     admin.setQuota(QuotaSettingsFactory.unthrottleUser(userName));
     triggerUserCacheRefresh(TEST_UTIL, true, TABLE_NAME);
-    assertEquals(100, doScans(100, table));
-    assertEquals(100, doScans(100, table));
+    testTraffic(() -> doScans(100, table), 100, 0);
+    testTraffic(() -> doScans(100, table), 100, 0);
   }
 
   @Test
@@ -187,34 +182,52 @@ public void testBBSMultiGet() throws Exception {
     waitMinuteQuota();
 
     // should execute 1 request
-    assertEquals(1, doMultiGets(10, 10, rowCount, FAMILY, QUALIFIER, table));
+    testTraffic(() -> doMultiGets(10, 10, rowCount, FAMILY, QUALIFIER, table), 1, 1);
 
     // Remove all the limits
     admin.setQuota(QuotaSettingsFactory.unthrottleUser(userName));
     triggerUserCacheRefresh(TEST_UTIL, true, TABLE_NAME);
-    assertEquals(100, doMultiGets(100, 10, rowCount, FAMILY, QUALIFIER, table));
-    assertEquals(100, doMultiGets(100, 10, rowCount, FAMILY, QUALIFIER, table));
+    testTraffic(() -> doMultiGets(100, 10, rowCount, FAMILY, QUALIFIER, table), 100, 0);
+    testTraffic(() -> doMultiGets(100, 10, rowCount, FAMILY, QUALIFIER, table), 100, 0);
 
     // Add ~100 block/min limit
     admin.setQuota(QuotaSettingsFactory.throttleUser(userName, ThrottleType.REQUEST_SIZE,
       Math.round(100.1 * blockSize), TimeUnit.MINUTES));
     triggerUserCacheRefresh(TEST_UTIL, false, TABLE_NAME);
 
     // should execute approximately 10 batches of 10 requests
-    long successfulMultiGets = doMultiGets(20, 10, rowCount, FAMILY, QUALIFIER, table);
-    assertTrue(successfulMultiGets >= 9);
-    assertTrue(successfulMultiGets <= 11);
+    testTraffic(() -> doMultiGets(20, 10, rowCount, FAMILY, QUALIFIER, table), 10, 1);
 
     // wait a minute and you should get another ~10 batches of 10 requests
     waitMinuteQuota();
-    successfulMultiGets = doMultiGets(20, 10, rowCount, FAMILY, QUALIFIER, table);
-    assertTrue(successfulMultiGets >= 9);
-    assertTrue(successfulMultiGets <= 11);
+    testTraffic(() -> doMultiGets(20, 10, rowCount, FAMILY, QUALIFIER, table), 10, 1);
 
     // Remove all the limits
     admin.setQuota(QuotaSettingsFactory.unthrottleUser(userName));
     triggerUserCacheRefresh(TEST_UTIL, true, TABLE_NAME);
-    assertEquals(100, doMultiGets(100, 10, rowCount, FAMILY, QUALIFIER, table));
-    assertEquals(100, doMultiGets(100, 10, rowCount, FAMILY, QUALIFIER, table));
+    testTraffic(() -> doMultiGets(100, 10, rowCount, FAMILY, QUALIFIER, table), 100, 0);
+    testTraffic(() -> doMultiGets(100, 10, rowCount, FAMILY, QUALIFIER, table), 100, 0);
+  }
+
+  private void testTraffic(Callable<Long> trafficCallable, long expectedSuccess, long marginOfError)
+    throws Exception {
+    TEST_UTIL.waitFor(90_000, () -> {
+      long actualSuccess;
+      try {
+        actualSuccess = trafficCallable.call();
+      } catch (Exception e) {
+        throw new RuntimeException(e);
+      }
+      LOG.info("Traffic test yielded {} successful requests. Expected {} +/- {}", actualSuccess,
+        expectedSuccess, marginOfError);
+      boolean success = (actualSuccess >= expectedSuccess - marginOfError)
+        && (actualSuccess <= expectedSuccess + marginOfError);
+      if (!success) {
+        triggerUserCacheRefresh(TEST_UTIL, true, TABLE_NAME);
+        waitMinuteQuota();
+        Thread.sleep(15_000L);
+      }
+      return success;
+    });
   }
 }