HBASE-22012 Space Quota: DisableTableViolationPolicy will cause cycles of enable/disable table

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

@@ -25,6 +25,7 @@
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -694,23 +695,65 @@ public Map<RegionInfo, Long> snapshotRegionSizes() {
     return copy;
   }
 
-  int pruneEntriesOlderThan(long timeToPruneBefore) {
+  int pruneEntriesOlderThan(long timeToPruneBefore, QuotaObserverChore quotaObserverChore) {
     if (regionSizes == null) {
       return 0;
     }
     int numEntriesRemoved = 0;
-    Iterator<Entry<RegionInfo,SizeSnapshotWithTimestamp>> iterator =
+    Iterator<Entry<RegionInfo, SizeSnapshotWithTimestamp>> iterator =
         regionSizes.entrySet().iterator();
     while (iterator.hasNext()) {
-      long currentEntryTime = iterator.next().getValue().getTime();
-      if (currentEntryTime < timeToPruneBefore) {
+      RegionInfo regionInfo = iterator.next().getKey();
+      long currentEntryTime = regionSizes.get(regionInfo).getTime();
+      // do not prune the entries if table is in violation and
+      // violation policy is disable to avoid cycle of enable/disable.
+      // Please refer HBASE-22012 for more details.
+      // prune entries older than time.
+      if (currentEntryTime < timeToPruneBefore && !isInViolationAndPolicyDisable(
+          regionInfo.getTable(), quotaObserverChore)) {
         iterator.remove();
         numEntriesRemoved++;
       }
     }
     return numEntriesRemoved;
   }
 
+  /**
+   * Method to check if a table is in violation and policy set on table is DISABLE.
+   *
+   * @param tableName          tableName to check.
+   * @param quotaObserverChore QuotaObserverChore instance
+   * @return returns true if table is in violation and policy is disable else false.
+   */
+  private boolean isInViolationAndPolicyDisable(TableName tableName,
+      QuotaObserverChore quotaObserverChore) {
+    boolean isInViolationAtTable = false;
+    boolean isInViolationAtNamespace = false;
+    SpaceViolationPolicy policy = null;
+    // Get Current Snapshot for the given table
+    SpaceQuotaSnapshot tableQuotaSnapshot = quotaObserverChore.getTableQuotaSnapshot(tableName);
+    SpaceQuotaSnapshot namespaceQuotaSnapshot =
+        quotaObserverChore.getNamespaceQuotaSnapshot(tableName.getNamespaceAsString());
+    if (tableQuotaSnapshot != null) {
+      // check if table in violation
+      isInViolationAtTable = tableQuotaSnapshot.getQuotaStatus().isInViolation();
+      Optional<SpaceViolationPolicy> tablePolicy = tableQuotaSnapshot.getQuotaStatus().getPolicy();
+      if (tablePolicy.isPresent()) {
+        policy = tablePolicy.get();
+      }
+    } else if (namespaceQuotaSnapshot != null) {
+      // check namespace in violation
+      isInViolationAtNamespace = namespaceQuotaSnapshot.getQuotaStatus().isInViolation();
+      Optional<SpaceViolationPolicy> namespacePolicy =
+          tableQuotaSnapshot.getQuotaStatus().getPolicy();
+      if (namespacePolicy.isPresent()) {
+        policy = namespacePolicy.get();
+      }
+    }
+    return (policy == SpaceViolationPolicy.DISABLE) && (isInViolationAtTable
+        || isInViolationAtNamespace);
+  }
+
   public void processFileArchivals(FileArchiveNotificationRequest request, Connection conn,
       Configuration conf, FileSystem fs) throws IOException {
     final HashMultimap<TableName,Entry<String,Long>> archivedFilesByTable = HashMultimap.create();

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

@@ -471,7 +471,7 @@ void updateNamespaceQuota(
   void pruneOldRegionReports() {
     final long now = EnvironmentEdgeManager.currentTime();
     final long pruneTime = now - regionReportLifetimeMillis;
-    final int numRemoved = quotaManager.pruneEntriesOlderThan(pruneTime);
+    final int numRemoved = quotaManager.pruneEntriesOlderThan(pruneTime,this);
     if (LOG.isTraceEnabled()) {
       LOG.trace("Removed " + numRemoved + " old region size reports that were older than "
           + pruneTime + ".");

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

@@ -73,18 +73,19 @@ public void testOldEntriesRemoved() {
 
     assertEquals(5, manager.snapshotRegionSizes().size());
 
+    QuotaObserverChore chore = mock(QuotaObserverChore.class);
     // Prune nothing
-    assertEquals(0, manager.pruneEntriesOlderThan(0));
+    assertEquals(0, manager.pruneEntriesOlderThan(0, chore));
     assertEquals(5, manager.snapshotRegionSizes().size());
-    assertEquals(0, manager.pruneEntriesOlderThan(10));
+    assertEquals(0, manager.pruneEntriesOlderThan(10, chore));
     assertEquals(5, manager.snapshotRegionSizes().size());
 
     // Prune the elements at time1
-    assertEquals(2, manager.pruneEntriesOlderThan(15));
+    assertEquals(2, manager.pruneEntriesOlderThan(15, chore));
     assertEquals(3, manager.snapshotRegionSizes().size());
 
     // Prune the elements at time2
-    assertEquals(2, manager.pruneEntriesOlderThan(30));
+    assertEquals(2, manager.pruneEntriesOlderThan(30, chore));
     assertEquals(1, manager.snapshotRegionSizes().size());
   }
 }

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

@@ -19,24 +19,31 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.DoNotRetryIOException;
 import org.apache.hadoop.hbase.HBaseClassTestRule;
 import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.MetaTableAccessor;
 import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.Waiter;
 import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.Append;
 import org.apache.hadoop.hbase.client.Delete;
 import org.apache.hadoop.hbase.client.Increment;
 import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.master.HMaster;
 import org.apache.hadoop.hbase.security.AccessDeniedException;
 import org.apache.hadoop.hbase.testclassification.LargeTests;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.util.StringUtils;
 import org.junit.AfterClass;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
@@ -221,4 +228,32 @@ public void testTableQuotaOverridesNamespaceQuota() throws Exception {
         Bytes.toBytes("reject"));
     helper.verifyViolation(policy, tn, p);
   }
+
+  @Test
+  public void testDisablePolicyQuotaAndViolate() throws Exception {
+    TableName tableName = helper.createTable();
+    helper.setQuotaLimit(tableName, SpaceViolationPolicy.DISABLE, 1L);
+    helper.writeData(tableName, SpaceQuotaHelperForTests.ONE_MEGABYTE * 2L);
+    TEST_UTIL.getConfiguration()
+        .setLong("hbase.master.quotas.region.report.retention.millis", 100);
+
+    HMaster master = TEST_UTIL.getMiniHBaseCluster().getMaster();
+    MasterQuotaManager quotaManager = master.getMasterQuotaManager();
+
+    // Make sure the master has report for the table.
+    Waiter.waitFor(TEST_UTIL.getConfiguration(), 30 * 1000, new Waiter.Predicate<Exception>() {
+      @Override
+      public boolean evaluate() throws Exception {
+        Map<RegionInfo, Long> regionSizes = quotaManager.snapshotRegionSizes();
+        List<RegionInfo> tableRegions =
+            MetaTableAccessor.getTableRegions(TEST_UTIL.getConnection(), tableName);
+        return regionSizes.containsKey(tableRegions.get(0));
+      }
+    });
+
+    // Check if disabled table region report present in the map after retention period expired.
+    // It should be present after retention period expired.
+    Assert.assertTrue(quotaManager.snapshotRegionSizes().keySet().stream()
+        .filter(k -> k.getTable().equals(tableName)).count() > 0);
+  }
 }