http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/SpaceViolationPolicyEnforcementFactory.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/SpaceViolationPolicyEnforcementFactory.java
 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/SpaceViolationPolicyEnforcementFactory.java
new file mode 100644
index 0000000..6b754b9
--- /dev/null
+++ 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/SpaceViolationPolicyEnforcementFactory.java
@@ -0,0 +1,95 @@
+/*
+ * 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 org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.classification.InterfaceStability;
+import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshot.SpaceQuotaStatus;
+import 
org.apache.hadoop.hbase.quotas.policies.BulkLoadVerifyingViolationPolicyEnforcement;
+import 
org.apache.hadoop.hbase.quotas.policies.DisableTableViolationPolicyEnforcement;
+import 
org.apache.hadoop.hbase.quotas.policies.NoInsertsViolationPolicyEnforcement;
+import 
org.apache.hadoop.hbase.quotas.policies.NoWritesCompactionsViolationPolicyEnforcement;
+import 
org.apache.hadoop.hbase.quotas.policies.NoWritesViolationPolicyEnforcement;
+import org.apache.hadoop.hbase.regionserver.RegionServerServices;
+
+/**
+ * A factory class for instantiating {@link SpaceViolationPolicyEnforcement} 
instances.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class SpaceViolationPolicyEnforcementFactory {
+
+  private static final SpaceViolationPolicyEnforcementFactory INSTANCE =
+      new SpaceViolationPolicyEnforcementFactory();
+
+  private SpaceViolationPolicyEnforcementFactory() {}
+
+  /**
+   * Returns an instance of this factory.
+   */
+  public static SpaceViolationPolicyEnforcementFactory getInstance() {
+    return INSTANCE;
+  }
+
+  /**
+   * Constructs the appropriate {@link SpaceViolationPolicyEnforcement} for 
tables that are
+   * in violation of their space quota.
+   */
+  public SpaceViolationPolicyEnforcement create(
+      RegionServerServices rss, TableName tableName, SpaceQuotaSnapshot 
snapshot) {
+    SpaceViolationPolicyEnforcement enforcement;
+    SpaceQuotaStatus status = snapshot.getQuotaStatus();
+    if (!status.isInViolation()) {
+      throw new IllegalArgumentException(tableName + " is not in violation. 
Snapshot=" + snapshot);
+    }
+    switch (status.getPolicy()) {
+      case DISABLE:
+        enforcement = new DisableTableViolationPolicyEnforcement();
+        break;
+      case NO_WRITES_COMPACTIONS:
+        enforcement = new NoWritesCompactionsViolationPolicyEnforcement();
+        break;
+      case NO_WRITES:
+        enforcement = new NoWritesViolationPolicyEnforcement();
+        break;
+      case NO_INSERTS:
+        enforcement = new NoInsertsViolationPolicyEnforcement();
+        break;
+      default:
+        throw new IllegalArgumentException("Unhandled SpaceViolationPolicy: " 
+ status.getPolicy());
+    }
+    enforcement.initialize(rss, tableName, snapshot);
+    return enforcement;
+  }
+
+  /**
+   * Creates the "default" {@link SpaceViolationPolicyEnforcement} for a table 
that isn't in
+   * violation. This is used to have uniform policy checking for tables in and 
not quotas.
+   */
+  public SpaceViolationPolicyEnforcement createWithoutViolation(
+      RegionServerServices rss, TableName tableName, SpaceQuotaSnapshot 
snapshot) {
+    SpaceQuotaStatus status = snapshot.getQuotaStatus();
+    if (status.isInViolation()) {
+      throw new IllegalArgumentException(
+          tableName + " is in violation. Logic error. Snapshot=" + snapshot);
+    }
+    BulkLoadVerifyingViolationPolicyEnforcement enforcement = new 
BulkLoadVerifyingViolationPolicyEnforcement();
+    enforcement.initialize(rss, tableName, snapshot);
+    return enforcement;
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TableQuotaSnapshotStore.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TableQuotaSnapshotStore.java
 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TableQuotaSnapshotStore.java
new file mode 100644
index 0000000..e196354
--- /dev/null
+++ 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TableQuotaSnapshotStore.java
@@ -0,0 +1,127 @@
+/*
+ * 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.io.IOException;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
+import java.util.Map.Entry;
+
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshot.SpaceQuotaStatus;
+import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.Quotas;
+import 
org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.SpaceQuota;
+
+import com.google.common.base.Predicate;
+import com.google.common.collect.Iterables;
+
+/**
+ * {@link QuotaSnapshotStore} for tables.
+ */
+@InterfaceAudience.Private
+public class TableQuotaSnapshotStore implements QuotaSnapshotStore<TableName> {
+  private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+  private final ReadLock rlock = lock.readLock();
+  private final WriteLock wlock = lock.writeLock();
+
+  private final Connection conn;
+  private final QuotaObserverChore chore;
+  private Map<HRegionInfo,Long> regionUsage;
+
+  public TableQuotaSnapshotStore(Connection conn, QuotaObserverChore chore, 
Map<HRegionInfo,Long> regionUsage) {
+    this.conn = Objects.requireNonNull(conn);
+    this.chore = Objects.requireNonNull(chore);
+    this.regionUsage = Objects.requireNonNull(regionUsage);
+  }
+
+  @Override
+  public SpaceQuota getSpaceQuota(TableName subject) throws IOException {
+    Quotas quotas = getQuotaForTable(subject);
+    if (null != quotas && quotas.hasSpace()) {
+      return quotas.getSpace();
+    }
+    return null;
+  }
+  /**
+   * Fetches the table quota. Visible for mocking/testing.
+   */
+  Quotas getQuotaForTable(TableName table) throws IOException {
+    return QuotaTableUtil.getTableQuota(conn, table);
+  }
+
+  @Override
+  public SpaceQuotaSnapshot getCurrentState(TableName table) {
+    // Defer the "current state" to the chore
+    return chore.getTableQuotaSnapshot(table);
+  }
+
+  @Override
+  public SpaceQuotaSnapshot getTargetState(TableName table, SpaceQuota 
spaceQuota) {
+    rlock.lock();
+    try {
+      final long sizeLimitInBytes = spaceQuota.getSoftLimit();
+      long sum = 0L;
+      for (Entry<HRegionInfo,Long> entry : filterBySubject(table)) {
+        sum += entry.getValue();
+      }
+      // Observance is defined as the size of the table being less than the 
limit
+      SpaceQuotaStatus status = sum <= sizeLimitInBytes ? 
SpaceQuotaStatus.notInViolation()
+          : new 
SpaceQuotaStatus(ProtobufUtil.toViolationPolicy(spaceQuota.getViolationPolicy()));
+      return new SpaceQuotaSnapshot(status, sum, sizeLimitInBytes);
+    } finally {
+      rlock.unlock();
+    }
+  }
+
+  @Override
+  public Iterable<Entry<HRegionInfo,Long>> filterBySubject(TableName table) {
+    rlock.lock();
+    try {
+      return Iterables.filter(regionUsage.entrySet(), new 
Predicate<Entry<HRegionInfo,Long>>() {
+        @Override
+        public boolean apply(Entry<HRegionInfo,Long> input) {
+          return table.equals(input.getKey().getTable());
+        }
+      });
+    } finally {
+      rlock.unlock();
+    }
+  }
+
+  @Override
+  public void setCurrentState(TableName table, SpaceQuotaSnapshot snapshot) {
+    // Defer the "current state" to the chore
+    this.chore.setTableQuotaViolation(table, snapshot);
+  }
+
+  @Override
+  public void setRegionUsage(Map<HRegionInfo,Long> regionUsage) {
+    wlock.lock();
+    try {
+      this.regionUsage = Objects.requireNonNull(regionUsage);
+    } finally {
+      wlock.unlock();
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TableQuotaViolationStore.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TableQuotaViolationStore.java
 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TableQuotaViolationStore.java
deleted file mode 100644
index 6aba1cf..0000000
--- 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TableQuotaViolationStore.java
+++ /dev/null
@@ -1,127 +0,0 @@
-/*
- * 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.io.IOException;
-import java.util.Map;
-import java.util.Objects;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
-import java.util.Map.Entry;
-
-import org.apache.hadoop.hbase.HRegionInfo;
-import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.classification.InterfaceAudience;
-import org.apache.hadoop.hbase.client.Connection;
-import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.Quotas;
-import 
org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.SpaceQuota;
-
-import com.google.common.base.Predicate;
-import com.google.common.collect.Iterables;
-
-/**
- * {@link QuotaViolationStore} for tables.
- */
-@InterfaceAudience.Private
-public class TableQuotaViolationStore implements 
QuotaViolationStore<TableName> {
-  private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
-  private final ReadLock rlock = lock.readLock();
-  private final WriteLock wlock = lock.writeLock();
-
-  private final Connection conn;
-  private final QuotaObserverChore chore;
-  private Map<HRegionInfo,Long> regionUsage;
-
-  public TableQuotaViolationStore(Connection conn, QuotaObserverChore chore, 
Map<HRegionInfo,Long> regionUsage) {
-    this.conn = Objects.requireNonNull(conn);
-    this.chore = Objects.requireNonNull(chore);
-    this.regionUsage = Objects.requireNonNull(regionUsage);
-  }
-
-  @Override
-  public SpaceQuota getSpaceQuota(TableName subject) throws IOException {
-    Quotas quotas = getQuotaForTable(subject);
-    if (null != quotas && quotas.hasSpace()) {
-      return quotas.getSpace();
-    }
-    return null;
-  }
-  /**
-   * Fetches the table quota. Visible for mocking/testing.
-   */
-  Quotas getQuotaForTable(TableName table) throws IOException {
-    return QuotaTableUtil.getTableQuota(conn, table);
-  }
-
-  @Override
-  public ViolationState getCurrentState(TableName table) {
-    // Defer the "current state" to the chore
-    return chore.getTableQuotaViolation(table);
-  }
-
-  @Override
-  public ViolationState getTargetState(TableName table, SpaceQuota spaceQuota) 
{
-    rlock.lock();
-    try {
-      final long sizeLimitInBytes = spaceQuota.getSoftLimit();
-      long sum = 0L;
-      for (Entry<HRegionInfo,Long> entry : filterBySubject(table)) {
-        sum += entry.getValue();
-        if (sum > sizeLimitInBytes) {
-          // Short-circuit early
-          return ViolationState.IN_VIOLATION;
-        }
-      }
-      // Observance is defined as the size of the table being less than the 
limit
-      return sum <= sizeLimitInBytes ? ViolationState.IN_OBSERVANCE : 
ViolationState.IN_VIOLATION;
-    } finally {
-      rlock.unlock();
-    }
-  }
-
-  @Override
-  public Iterable<Entry<HRegionInfo,Long>> filterBySubject(TableName table) {
-    rlock.lock();
-    try {
-      return Iterables.filter(regionUsage.entrySet(), new 
Predicate<Entry<HRegionInfo,Long>>() {
-        @Override
-        public boolean apply(Entry<HRegionInfo,Long> input) {
-          return table.equals(input.getKey().getTable());
-        }
-      });
-    } finally {
-      rlock.unlock();
-    }
-  }
-
-  @Override
-  public void setCurrentState(TableName table, ViolationState state) {
-    // Defer the "current state" to the chore
-    this.chore.setTableQuotaViolation(table, state);
-  }
-
-  @Override
-  public void setRegionUsage(Map<HRegionInfo,Long> regionUsage) {
-    wlock.lock();
-    try {
-      this.regionUsage = Objects.requireNonNull(regionUsage);
-    } finally {
-      wlock.unlock();
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TableSpaceQuotaSnapshotNotifier.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TableSpaceQuotaSnapshotNotifier.java
 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TableSpaceQuotaSnapshotNotifier.java
new file mode 100644
index 0000000..548faf8
--- /dev/null
+++ 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TableSpaceQuotaSnapshotNotifier.java
@@ -0,0 +1,52 @@
+/*
+ * 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.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Table;
+
+/**
+ * A {@link SpaceQuotaSnapshotNotifier} which uses the hbase:quota table.
+ */
+public class TableSpaceQuotaSnapshotNotifier implements 
SpaceQuotaSnapshotNotifier {
+  private static final Log LOG = 
LogFactory.getLog(TableSpaceQuotaSnapshotNotifier.class);
+
+  private Connection conn;
+
+  @Override
+  public void transitionTable(
+      TableName tableName, SpaceQuotaSnapshot snapshot) throws IOException {
+    final Put p = QuotaTableUtil.createPutSpaceSnapshot(tableName, snapshot);
+    try (Table quotaTable = conn.getTable(QuotaTableUtil.QUOTA_TABLE_NAME)) {
+      if (LOG.isTraceEnabled()) {
+        LOG.trace("Persisting a space quota snapshot " + snapshot + " for " + 
tableName);
+      }
+      quotaTable.put(p);
+    }
+  }
+
+  @Override
+  public void initialize(Connection conn) {
+    this.conn = conn;
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TableSpaceQuotaViolationNotifier.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TableSpaceQuotaViolationNotifier.java
 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TableSpaceQuotaViolationNotifier.java
deleted file mode 100644
index a8b1c55..0000000
--- 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TableSpaceQuotaViolationNotifier.java
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * 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.io.IOException;
-
-import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.client.Connection;
-import org.apache.hadoop.hbase.client.Delete;
-import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.client.Table;
-
-/**
- * A {@link SpaceQuotaViolationNotifier} which uses the hbase:quota table.
- */
-public class TableSpaceQuotaViolationNotifier implements 
SpaceQuotaViolationNotifier {
-
-  private Connection conn;
-
-  @Override
-  public void transitionTableToViolation(
-      TableName tableName, SpaceViolationPolicy violationPolicy) throws 
IOException {
-    final Put p = QuotaTableUtil.createEnableViolationPolicyUpdate(tableName, 
violationPolicy);
-    try (Table quotaTable = conn.getTable(QuotaTableUtil.QUOTA_TABLE_NAME)) {
-      quotaTable.put(p);
-    }
-  }
-
-  @Override
-  public void transitionTableToObservance(TableName tableName) throws 
IOException {
-    final Delete d = 
QuotaTableUtil.createRemoveViolationPolicyUpdate(tableName);
-    try (Table quotaTable = conn.getTable(QuotaTableUtil.QUOTA_TABLE_NAME)) {
-      quotaTable.delete(d);
-    }
-  }
-
-  @Override
-  public void initialize(Connection conn) {
-    this.conn = conn;
-  }
-}

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/AbstractViolationPolicyEnforcement.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/AbstractViolationPolicyEnforcement.java
 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/AbstractViolationPolicyEnforcement.java
new file mode 100644
index 0000000..2d34d45
--- /dev/null
+++ 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/AbstractViolationPolicyEnforcement.java
@@ -0,0 +1,118 @@
+/*
+ * 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.policies;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.classification.InterfaceStability;
+import org.apache.hadoop.hbase.quotas.SpaceLimitingException;
+import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshot;
+import org.apache.hadoop.hbase.quotas.SpaceViolationPolicyEnforcement;
+import org.apache.hadoop.hbase.regionserver.RegionServerServices;
+
+/**
+ * Abstract implementation for {@link SpaceViolationPolicyEnforcement}.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public abstract class AbstractViolationPolicyEnforcement
+    implements SpaceViolationPolicyEnforcement {
+
+  RegionServerServices rss;
+  TableName tableName;
+  SpaceQuotaSnapshot quotaSnapshot;
+
+  public void setRegionServerServices(RegionServerServices rss) {
+    this.rss = Objects.requireNonNull(rss);
+  }
+
+  public void setTableName(TableName tableName) {
+    this.tableName = tableName;
+  }
+
+  public RegionServerServices getRegionServerServices() {
+    return this.rss;
+  }
+
+  public TableName getTableName() {
+    return this.tableName;
+  }
+
+  public void setQuotaSnapshot(SpaceQuotaSnapshot snapshot) {
+    this.quotaSnapshot = Objects.requireNonNull(snapshot);
+  }
+
+  @Override
+  public SpaceQuotaSnapshot getQuotaSnapshot() {
+    return this.quotaSnapshot;
+  }
+
+  @Override
+  public void initialize(RegionServerServices rss, TableName tableName, 
SpaceQuotaSnapshot snapshot) {
+    setRegionServerServices(rss);
+    setTableName(tableName);
+    setQuotaSnapshot(snapshot);
+  }
+
+  @Override
+  public boolean areCompactionsDisabled() {
+    return false;
+  }
+
+  @Override
+  public boolean shouldCheckBulkLoads() {
+    // Reference check. The singleton is used when no quota exists to check 
against
+    return SpaceQuotaSnapshot.getNoSuchSnapshot() != quotaSnapshot;
+  }
+
+  @Override
+  public void checkBulkLoad(FileSystem fs, List<String> paths) throws 
SpaceLimitingException {
+    // Compute the amount of space that could be used to save some arithmetic 
in the for-loop
+    final long sizeAvailableForBulkLoads = quotaSnapshot.getLimit() - 
quotaSnapshot.getUsage();
+    long size = 0L;
+    for (String path : paths) {
+      size += addSingleFile(fs, path);
+      if (size > sizeAvailableForBulkLoads) {
+        break;
+      }
+    }
+    if (size > sizeAvailableForBulkLoads) {
+      throw new SpaceLimitingException(getPolicyName(), "Bulk load of " + paths
+          + " is disallowed because the file(s) exceed the limits of a space 
quota.");
+    }
+  }
+
+  private long addSingleFile(FileSystem fs, String path) throws 
SpaceLimitingException {
+    final FileStatus status;
+    try {
+      status = fs.getFileStatus(new Path(Objects.requireNonNull(path)));
+    } catch (IOException e) {
+      throw new SpaceLimitingException(getPolicyName(), "Could not verify 
length of file to bulk load", e);
+    }
+    if (!status.isFile()) {
+      throw new IllegalArgumentException(path + " is not a file.");
+    }
+    return status.getLen();
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/BulkLoadVerifyingViolationPolicyEnforcement.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/BulkLoadVerifyingViolationPolicyEnforcement.java
 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/BulkLoadVerifyingViolationPolicyEnforcement.java
new file mode 100644
index 0000000..e4171ad
--- /dev/null
+++ 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/BulkLoadVerifyingViolationPolicyEnforcement.java
@@ -0,0 +1,50 @@
+/*
+ * 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.policies;
+
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.quotas.SpaceLimitingException;
+import org.apache.hadoop.hbase.quotas.SpaceViolationPolicyEnforcement;
+
+/**
+ * A {@link SpaceViolationPolicyEnforcement} instance which only checks for 
bulk loads. Useful for tables
+ * which have no violation policy. This is the default case for tables, as we 
want to make sure that
+ * a single bulk load call would violate the quota.
+ */
+@InterfaceAudience.Private
+public class BulkLoadVerifyingViolationPolicyEnforcement extends 
AbstractViolationPolicyEnforcement {
+
+  @Override
+  public void enable() {}
+
+  @Override
+  public void disable() {}
+
+  @Override
+  public String getPolicyName() {
+    return "BulkLoadVerifying";
+  }
+
+  @Override
+  public boolean areCompactionsDisabled() {
+    return false;
+  }
+
+  @Override
+  public void check(Mutation m) throws SpaceLimitingException {}
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/DisableTableViolationPolicyEnforcement.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/DisableTableViolationPolicyEnforcement.java
 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/DisableTableViolationPolicyEnforcement.java
new file mode 100644
index 0000000..0d6d886
--- /dev/null
+++ 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/DisableTableViolationPolicyEnforcement.java
@@ -0,0 +1,80 @@
+/*
+ * 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.policies;
+
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.TableNotDisabledException;
+import org.apache.hadoop.hbase.TableNotEnabledException;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.quotas.SpaceLimitingException;
+import org.apache.hadoop.hbase.quotas.SpaceViolationPolicy;
+import org.apache.hadoop.hbase.quotas.SpaceViolationPolicyEnforcement;
+
+/**
+ * A {@link SpaceViolationPolicyEnforcement} which disables the table. The 
enforcement
+ * counterpart to {@link SpaceViolationPolicy#DISABLE}.
+ */
+@InterfaceAudience.Private
+public class DisableTableViolationPolicyEnforcement extends 
AbstractViolationPolicyEnforcement {
+  private static final Log LOG = 
LogFactory.getLog(DisableTableViolationPolicyEnforcement.class);
+
+  @Override
+  public void enable() throws IOException {
+    try {
+      if (LOG.isTraceEnabled()) {
+        LOG.trace("Starting disable of " + getTableName());
+      }
+      
getRegionServerServices().getClusterConnection().getAdmin().disableTable(getTableName());
+      if (LOG.isTraceEnabled()) {
+        LOG.trace("Disable is complete for " + getTableName());
+      }
+    } catch (TableNotEnabledException tnee) {
+      // The state we wanted it to be in.
+    }
+  }
+
+  @Override
+  public void disable() throws IOException {
+    try {
+      if (LOG.isTraceEnabled()) {
+        LOG.trace("Starting enable of " + getTableName());
+      }
+      
getRegionServerServices().getClusterConnection().getAdmin().enableTable(getTableName());
+      if (LOG.isTraceEnabled()) {
+        LOG.trace("Enable is complete for " + getTableName());
+      }
+    } catch (TableNotDisabledException tnde) {
+      // The state we wanted it to be in
+    }
+  }
+
+  @Override
+  public void check(Mutation m) throws SpaceLimitingException {
+    // If this policy is enacted, then the table is (or should be) disabled.
+    throw new SpaceLimitingException(
+        getPolicyName(), "This table is disabled due to violating a space 
quota.");
+  }
+
+  @Override
+  public String getPolicyName() {
+    return SpaceViolationPolicy.DISABLE.name();
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/NoInsertsViolationPolicyEnforcement.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/NoInsertsViolationPolicyEnforcement.java
 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/NoInsertsViolationPolicyEnforcement.java
new file mode 100644
index 0000000..a60cb45
--- /dev/null
+++ 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/NoInsertsViolationPolicyEnforcement.java
@@ -0,0 +1,55 @@
+/*
+ * 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.policies;
+
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.client.Append;
+import org.apache.hadoop.hbase.client.Increment;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.quotas.SpaceLimitingException;
+import org.apache.hadoop.hbase.quotas.SpaceViolationPolicy;
+import org.apache.hadoop.hbase.quotas.SpaceViolationPolicyEnforcement;
+
+/**
+ * A {@link SpaceViolationPolicyEnforcement} which disallows any inserts to 
the table. The
+ * enforcement counterpart to {@link SpaceViolationPolicy#NO_INSERTS}.
+ */
+@InterfaceAudience.Private
+public class NoInsertsViolationPolicyEnforcement extends 
AbstractViolationPolicyEnforcement {
+
+  @Override
+  public void enable() {}
+
+  @Override
+  public void disable() {}
+
+  @Override
+  public void check(Mutation m) throws SpaceLimitingException {
+    // Disallow all "new" data flowing into HBase, but allow Deletes (even 
though we know they will
+    // temporarily increase utilization).
+    if (m instanceof Append  || m instanceof Increment || m instanceof Put) {
+      throw new SpaceLimitingException(getPolicyName(),
+          m.getClass().getSimpleName() + "s are disallowed due to a space 
quota.");
+    }
+  }
+
+  @Override
+  public String getPolicyName() {
+    return SpaceViolationPolicy.NO_INSERTS.name();
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/NoWritesCompactionsViolationPolicyEnforcement.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/NoWritesCompactionsViolationPolicyEnforcement.java
 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/NoWritesCompactionsViolationPolicyEnforcement.java
new file mode 100644
index 0000000..e7f872c
--- /dev/null
+++ 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/NoWritesCompactionsViolationPolicyEnforcement.java
@@ -0,0 +1,64 @@
+/*
+ * 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.policies;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.quotas.SpaceViolationPolicy;
+import org.apache.hadoop.hbase.quotas.SpaceViolationPolicyEnforcement;
+
+/**
+ * A {@link SpaceViolationPolicyEnforcement} implementation which disables all 
updates and
+ * compactions. The enforcement counterpart to {@link 
SpaceViolationPolicy#NO_WRITES_COMPACTIONS}.
+ */
+@InterfaceAudience.Private
+public class NoWritesCompactionsViolationPolicyEnforcement
+    extends NoWritesViolationPolicyEnforcement {
+  private static final Log LOG = LogFactory.getLog(
+      NoWritesCompactionsViolationPolicyEnforcement.class);
+
+  private AtomicBoolean disableCompactions = new AtomicBoolean(false);
+
+  @Override
+  public synchronized void enable() {
+    boolean ret = disableCompactions.compareAndSet(false, true);
+    if (!ret && LOG.isTraceEnabled()) {
+      LOG.trace("Compactions were already disabled upon enabling the policy");
+    }
+  }
+
+  @Override
+  public synchronized void disable() {
+    boolean ret = disableCompactions.compareAndSet(true, false);
+    if (!ret && LOG.isTraceEnabled()) {
+      LOG.trace("Compactions were already enabled upon disabling the policy");
+    }
+  }
+
+  @Override
+  public String getPolicyName() {
+    return SpaceViolationPolicy.NO_WRITES_COMPACTIONS.name();
+  }
+
+  @Override
+  public boolean areCompactionsDisabled() {
+    return disableCompactions.get();
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/NoWritesViolationPolicyEnforcement.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/NoWritesViolationPolicyEnforcement.java
 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/NoWritesViolationPolicyEnforcement.java
new file mode 100644
index 0000000..a04f418
--- /dev/null
+++ 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/policies/NoWritesViolationPolicyEnforcement.java
@@ -0,0 +1,54 @@
+/*
+ * 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.policies;
+
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+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.Mutation;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.quotas.SpaceLimitingException;
+import org.apache.hadoop.hbase.quotas.SpaceViolationPolicy;
+import org.apache.hadoop.hbase.quotas.SpaceViolationPolicyEnforcement;
+
+/**
+ * A {@link SpaceViolationPolicyEnforcement} implementation which disables all 
writes flowing
+ * into HBase. The enforcement counterpart to {@link 
SpaceViolationPolicy#NO_WRITES}.
+ */
+@InterfaceAudience.Private
+public class NoWritesViolationPolicyEnforcement extends 
AbstractViolationPolicyEnforcement {
+
+  @Override
+  public void enable() {}
+
+  @Override
+  public void disable() {}
+
+  @Override
+  public void check(Mutation m) throws SpaceLimitingException {
+    if (m instanceof Append || m instanceof Delete || m instanceof Increment 
|| m instanceof Put) {
+      throw new SpaceLimitingException(getPolicyName(),
+          m.getClass().getSimpleName() + "s are disallowed due to a space 
quota.");
+    }
+  }
+
+  @Override
+  public String getPolicyName() {
+    return SpaceViolationPolicy.NO_WRITES.name();
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactSplitThread.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactSplitThread.java
 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactSplitThread.java
index eba984a..9aa0042 100644
--- 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactSplitThread.java
+++ 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactSplitThread.java
@@ -38,6 +38,7 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.classification.InterfaceAudience;
 import org.apache.hadoop.hbase.conf.ConfigurationManager;
 import org.apache.hadoop.hbase.conf.PropagatingConfigurationObserver;
+import org.apache.hadoop.hbase.quotas.RegionServerSpaceQuotaManager;
 import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
 import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest;
 import 
org.apache.hadoop.hbase.regionserver.throttle.CompactionThroughputControllerFactory;
@@ -337,6 +338,17 @@ public class CompactSplitThread implements 
CompactionRequestor, PropagatingConfi
       if (compaction == null) return null; // message logged inside
     }
 
+    final RegionServerSpaceQuotaManager spaceQuotaManager =
+      this.server.getRegionServerSpaceQuotaManager();
+    if (null != spaceQuotaManager && spaceQuotaManager.areCompactionsDisabled(
+        r.getTableDesc().getTableName())) {
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Ignoring compaction request for " + r + " as an active 
space quota violation "
+            + " policy disallows compactions.");
+      }
+      return null;
+    }
+
     // We assume that most compactions are small. So, put system compactions 
into small
     // pool; we will do selection there, and move to large pool if necessary.
     ThreadPoolExecutor pool = (selectNow && 
s.throttleCompaction(compaction.getRequest().getSize()))

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
index 4ce43c9..fad21b0 100644
--- 
a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
+++ 
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
@@ -94,8 +94,12 @@ import org.apache.hadoop.hbase.ipc.RpcServerInterface;
 import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
 import org.apache.hadoop.hbase.ipc.ServerRpcController;
 import org.apache.hadoop.hbase.master.MasterRpcServices;
+import org.apache.hadoop.hbase.quotas.ActivePolicyEnforcement;
 import org.apache.hadoop.hbase.quotas.OperationQuota;
+import org.apache.hadoop.hbase.quotas.QuotaUtil;
 import org.apache.hadoop.hbase.quotas.RegionServerRpcQuotaManager;
+import org.apache.hadoop.hbase.quotas.RegionServerSpaceQuotaManager;
+import org.apache.hadoop.hbase.quotas.SpaceViolationPolicyEnforcement;
 import org.apache.hadoop.hbase.regionserver.HRegion.RegionScannerImpl;
 import org.apache.hadoop.hbase.regionserver.Leases.Lease;
 import org.apache.hadoop.hbase.regionserver.Leases.LeaseStillHeldException;
@@ -193,7 +197,6 @@ import 
org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.BulkLoadDescr
 import 
org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.CompactionDescriptor;
 import 
org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.FlushDescriptor;
 import 
org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.RegionEventDescriptor;
-import org.apache.hadoop.hbase.quotas.RegionServerRpcQuotaManager;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.DNS;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
@@ -559,8 +562,8 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
    */
   private boolean checkAndRowMutate(final Region region, final 
List<ClientProtos.Action> actions,
       final CellScanner cellScanner, byte[] row, byte[] family, byte[] 
qualifier,
-      CompareOp compareOp, ByteArrayComparable comparator,
-                                    RegionActionResult.Builder builder) throws 
IOException {
+      CompareOp compareOp, ByteArrayComparable comparator, 
RegionActionResult.Builder builder,
+      ActivePolicyEnforcement spaceQuotaEnforcement) throws IOException {
     if (!region.getRegionInfo().isMetaTable()) {
       regionServer.cacheFlusher.reclaimMemStoreMemory();
     }
@@ -579,10 +582,14 @@ public class RSRpcServices implements 
HBaseRPCErrorHandler,
       }
       switch (type) {
         case PUT:
-          rm.add(ProtobufUtil.toPut(action.getMutation(), cellScanner));
+          Put put = ProtobufUtil.toPut(action.getMutation(), cellScanner);
+          spaceQuotaEnforcement.getPolicyEnforcement(region).check(put);
+          rm.add(put);
           break;
         case DELETE:
-          rm.add(ProtobufUtil.toDelete(action.getMutation(), cellScanner));
+          Delete del = ProtobufUtil.toDelete(action.getMutation(), 
cellScanner);
+          spaceQuotaEnforcement.getPolicyEnforcement(region).check(del);
+          rm.add(del);
           break;
         default:
           throw new DoNotRetryIOException("Atomic put and/or delete only, not 
" + type.name());
@@ -609,10 +616,12 @@ public class RSRpcServices implements 
HBaseRPCErrorHandler,
    * @throws IOException
    */
   private Result append(final Region region, final OperationQuota quota,
-      final MutationProto mutation, final CellScanner cellScanner, long 
nonceGroup)
+      final MutationProto mutation, final CellScanner cellScanner, long 
nonceGroup,
+      ActivePolicyEnforcement spaceQuota)
       throws IOException {
     long before = EnvironmentEdgeManager.currentTime();
     Append append = ProtobufUtil.toAppend(mutation, cellScanner);
+    spaceQuota.getPolicyEnforcement(region).check(append);
     quota.addMutation(append);
     Result r = null;
     if (region.getCoprocessorHost() != null) {
@@ -657,10 +666,12 @@ public class RSRpcServices implements 
HBaseRPCErrorHandler,
    * @throws IOException
    */
   private Result increment(final Region region, final OperationQuota quota,
-      final MutationProto mutation, final CellScanner cells, long nonceGroup)
+      final MutationProto mutation, final CellScanner cells, long nonceGroup,
+      ActivePolicyEnforcement spaceQuota)
       throws IOException {
     long before = EnvironmentEdgeManager.currentTime();
     Increment increment = ProtobufUtil.toIncrement(mutation, cells);
+    spaceQuota.getPolicyEnforcement(region).check(increment);
     quota.addMutation(increment);
     Result r = null;
     if (region.getCoprocessorHost() != null) {
@@ -712,7 +723,8 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
   private List<CellScannable> doNonAtomicRegionMutation(final Region region,
       final OperationQuota quota, final RegionAction actions, final 
CellScanner cellScanner,
       final RegionActionResult.Builder builder, List<CellScannable> 
cellsToReturn, long nonceGroup,
-      final RegionScannersCloseCallBack closeCallBack, RpcCallContext context) 
{
+      final RegionScannersCloseCallBack closeCallBack, RpcCallContext context,
+      ActivePolicyEnforcement spaceQuotaEnforcement) {
     // Gather up CONTIGUOUS Puts and Deletes in this mutations List.  Idea is 
that rather than do
     // one at a time, we instead pass them in batch.  Be aware that the 
corresponding
     // ResultOrException instance that matches each Put or Delete is then 
added down in the
@@ -805,15 +817,17 @@ public class RSRpcServices implements 
HBaseRPCErrorHandler,
           if (type != MutationType.PUT && type != MutationType.DELETE && 
mutations != null &&
               !mutations.isEmpty()) {
             // Flush out any Puts or Deletes already collected.
-            doBatchOp(builder, region, quota, mutations, cellScanner);
+            doBatchOp(builder, region, quota, mutations, cellScanner, 
spaceQuotaEnforcement);
             mutations.clear();
           }
           switch (type) {
             case APPEND:
-              r = append(region, quota, action.getMutation(), cellScanner, 
nonceGroup);
+              r = append(region, quota, action.getMutation(), cellScanner, 
nonceGroup,
+                  spaceQuotaEnforcement);
               break;
             case INCREMENT:
-              r = increment(region, quota, action.getMutation(), cellScanner, 
nonceGroup);
+              r = increment(region, quota, action.getMutation(), cellScanner, 
nonceGroup,
+                  spaceQuotaEnforcement);
               break;
             case PUT:
             case DELETE:
@@ -864,7 +878,7 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
     }
     // Finish up any outstanding mutations
     if (mutations != null && !mutations.isEmpty()) {
-      doBatchOp(builder, region, quota, mutations, cellScanner);
+      doBatchOp(builder, region, quota, mutations, cellScanner, 
spaceQuotaEnforcement);
     }
     return cellsToReturn;
   }
@@ -878,7 +892,7 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
    */
   private void doBatchOp(final RegionActionResult.Builder builder, final 
Region region,
       final OperationQuota quota, final List<ClientProtos.Action> mutations,
-      final CellScanner cells) {
+      final CellScanner cells, ActivePolicyEnforcement spaceQuotaEnforcement) {
     Mutation[] mArray = new Mutation[mutations.size()];
     long before = EnvironmentEdgeManager.currentTime();
     boolean batchContainsPuts = false, batchContainsDelete = false;
@@ -895,6 +909,8 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
           batchContainsDelete = true;
         }
         mArray[i++] = mutation;
+        // Check if a space quota disallows this mutation
+        spaceQuotaEnforcement.getPolicyEnforcement(region).check(mutation);
         quota.addMutation(mutation);
       }
 
@@ -1265,10 +1281,14 @@ public class RSRpcServices implements 
HBaseRPCErrorHandler,
     return regionServer.getConfiguration();
   }
 
-  private RegionServerRpcQuotaManager getQuotaManager() {
+  private RegionServerRpcQuotaManager getRpcQuotaManager() {
     return regionServer.getRegionServerRpcQuotaManager();
   }
 
+  private RegionServerSpaceQuotaManager getSpaceQuotaManager() {
+    return regionServer.getRegionServerSpaceQuotaManager();
+  }
+
   void start() {
     rpcServer.start();
   }
@@ -1443,6 +1463,12 @@ public class RSRpcServices implements 
HBaseRPCErrorHandler,
       checkOpen();
       requestCount.increment();
       Region region = getRegion(request.getRegion());
+      if (QuotaUtil.isQuotaEnabled(getConfiguration()) &&
+          
this.regionServer.getRegionServerSpaceQuotaManager().areCompactionsDisabled(
+              region.getTableDesc().getTableName())) {
+        throw new DoNotRetryIOException("Compactions on this region are "
+            + "disabled due to a space quota violation.");
+      }
       region.startRegionOperation(Operation.COMPACT_REGION);
       LOG.info("Compacting " + region.getRegionInfo().getRegionNameAsString());
       boolean major = false;
@@ -2129,6 +2155,21 @@ public class RSRpcServices implements 
HBaseRPCErrorHandler,
       boolean loaded = false;
       Map<byte[], List<Path>> map = null;
 
+      // Check to see if this bulk load would exceed the space quota for this 
table
+      if (QuotaUtil.isQuotaEnabled(getConfiguration())) {
+        ActivePolicyEnforcement activeSpaceQuotas = 
getSpaceQuotaManager().getActiveEnforcements();
+        SpaceViolationPolicyEnforcement enforcement = 
activeSpaceQuotas.getPolicyEnforcement(region);
+        if (null != enforcement) {
+          // Bulk loads must still be atomic. We must enact all or none.
+          List<String> filePaths = new 
ArrayList<>(request.getFamilyPathCount());
+          for (FamilyPath familyPath : request.getFamilyPathList()) {
+            filePaths.add(familyPath.getPath());
+          }
+          // Check if the batch of files exceeds the current quota
+          enforcement.checkBulkLoad(regionServer.getFileSystem(), filePaths);
+        }
+      }
+
       if (!request.hasBulkToken()) {
         // Old style bulk load. This will not be supported in future releases
         List<Pair<byte[], String>> familyPaths = new 
ArrayList<>(request.getFamilyPathCount());
@@ -2257,7 +2298,7 @@ public class RSRpcServices implements 
HBaseRPCErrorHandler,
       Boolean existence = null;
       Result r = null;
       RpcCallContext context = RpcServer.getCurrentCall();
-      quota = getQuotaManager().checkQuota(region, 
OperationQuota.OperationType.GET);
+      quota = getRpcQuotaManager().checkQuota(region, 
OperationQuota.OperationType.GET);
 
       Get clientGet = ProtobufUtil.toGet(get);
       if (get.getExistenceOnly() && region.getCoprocessorHost() != null) {
@@ -2395,6 +2436,7 @@ public class RSRpcServices implements 
HBaseRPCErrorHandler,
     this.rpcMultiRequestCount.increment();
     Map<RegionSpecifier, ClientProtos.RegionLoadStats> regionStats = new 
HashMap<>(request
       .getRegionActionCount());
+    ActivePolicyEnforcement spaceQuotaEnforcement = 
getSpaceQuotaManager().getActiveEnforcements();
     for (RegionAction regionAction : request.getRegionActionList()) {
       this.requestCount.add(regionAction.getActionCount());
       OperationQuota quota;
@@ -2403,7 +2445,7 @@ public class RSRpcServices implements 
HBaseRPCErrorHandler,
       RegionSpecifier regionSpecifier = regionAction.getRegion();
       try {
         region = getRegion(regionSpecifier);
-        quota = getQuotaManager().checkQuota(region, 
regionAction.getActionList());
+        quota = getRpcQuotaManager().checkQuota(region, 
regionAction.getActionList());
       } catch (IOException e) {
         rpcServer.getMetrics().exception(e);
         
regionActionResultBuilder.setException(ResponseConverter.buildException(e));
@@ -2431,7 +2473,7 @@ public class RSRpcServices implements 
HBaseRPCErrorHandler,
                 ProtobufUtil.toComparator(condition.getComparator());
             processed = checkAndRowMutate(region, regionAction.getActionList(),
                   cellScanner, row, family, qualifier, compareOp,
-                  comparator, regionActionResultBuilder);
+                  comparator, regionActionResultBuilder, 
spaceQuotaEnforcement);
           } else {
             mutateRows(region, regionAction.getActionList(), cellScanner,
                 regionActionResultBuilder);
@@ -2452,7 +2494,8 @@ public class RSRpcServices implements 
HBaseRPCErrorHandler,
           context.setCallBack(closeCallBack);
         }
         cellsToReturn = doNonAtomicRegionMutation(region, quota, regionAction, 
cellScanner,
-            regionActionResultBuilder, cellsToReturn, nonceGroup, 
closeCallBack, context);
+            regionActionResultBuilder, cellsToReturn, nonceGroup, 
closeCallBack, context,
+            spaceQuotaEnforcement);
       }
       responseBuilder.addRegionActionResult(regionActionResultBuilder.build());
       quota.close();
@@ -2519,6 +2562,7 @@ public class RSRpcServices implements 
HBaseRPCErrorHandler,
     CellScanner cellScanner = controller != null ? controller.cellScanner() : 
null;
     OperationQuota quota = null;
     RpcCallContext context = RpcServer.getCurrentCall();
+    ActivePolicyEnforcement spaceQuotaEnforcement = null;
     // Clear scanner so we are not holding on to reference across call.
     if (controller != null) {
       controller.setCellScanner(null);
@@ -2538,19 +2582,22 @@ public class RSRpcServices implements 
HBaseRPCErrorHandler,
       Boolean processed = null;
       MutationType type = mutation.getMutateType();
 
-      quota = getQuotaManager().checkQuota(region, 
OperationQuota.OperationType.MUTATE);
+      quota = getRpcQuotaManager().checkQuota(region, 
OperationQuota.OperationType.MUTATE);
+      spaceQuotaEnforcement = getSpaceQuotaManager().getActiveEnforcements();
 
       switch (type) {
       case APPEND:
         // TODO: this doesn't actually check anything.
-        r = append(region, quota, mutation, cellScanner, nonceGroup);
+        r = append(region, quota, mutation, cellScanner, nonceGroup, 
spaceQuotaEnforcement);
         break;
       case INCREMENT:
         // TODO: this doesn't actually check anything.
-        r = increment(region, quota, mutation, cellScanner, nonceGroup);
+        r = increment(region, quota, mutation, cellScanner, nonceGroup, 
spaceQuotaEnforcement);
         break;
       case PUT:
         Put put = ProtobufUtil.toPut(mutation, cellScanner);
+        // Throws an exception when violated
+        spaceQuotaEnforcement.getPolicyEnforcement(region).check(put);
         quota.addMutation(put);
         if (request.hasCondition()) {
           Condition condition = request.getCondition();
@@ -2580,6 +2627,7 @@ public class RSRpcServices implements 
HBaseRPCErrorHandler,
         break;
       case DELETE:
         Delete delete = ProtobufUtil.toDelete(mutation, cellScanner);
+        spaceQuotaEnforcement.getPolicyEnforcement(region).check(delete);
         quota.addMutation(delete);
         if (request.hasCondition()) {
           Condition condition = request.getCondition();
@@ -2998,7 +3046,7 @@ public class RSRpcServices implements 
HBaseRPCErrorHandler,
     }
     OperationQuota quota;
     try {
-      quota = getQuotaManager().checkQuota(region, 
OperationQuota.OperationType.SCAN);
+      quota = getRpcQuotaManager().checkQuota(region, 
OperationQuota.OperationType.SCAN);
     } catch (IOException e) {
       addScannerLeaseBack(lease);
       throw new ServiceException(e);

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/SpaceQuotaHelperForTests.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/SpaceQuotaHelperForTests.java
 
b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/SpaceQuotaHelperForTests.java
new file mode 100644
index 0000000..888978d
--- /dev/null
+++ 
b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/SpaceQuotaHelperForTests.java
@@ -0,0 +1,228 @@
+/*
+ * 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 static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.NamespaceDescriptor;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.rules.TestName;
+
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
+
+@InterfaceAudience.Private
+public class SpaceQuotaHelperForTests {
+  private static final Log LOG = 
LogFactory.getLog(SpaceQuotaHelperForTests.class);
+
+  public static final int SIZE_PER_VALUE = 256;
+  public static final String F1 = "f1";
+  public static final long ONE_KILOBYTE = 1024L;
+  public static final long ONE_MEGABYTE = ONE_KILOBYTE * ONE_KILOBYTE;
+
+  private final HBaseTestingUtility testUtil;
+  private final TestName testName;
+  private final AtomicLong counter;
+
+  public SpaceQuotaHelperForTests(
+      HBaseTestingUtility testUtil, TestName testName, AtomicLong counter) {
+    this.testUtil = Objects.requireNonNull(testUtil);
+    this.testName = Objects.requireNonNull(testName);
+    this.counter = Objects.requireNonNull(counter);
+  }
+
+  //
+  // Helpers
+  //
+
+  void writeData(TableName tn, long sizeInBytes) throws IOException {
+    final Connection conn = testUtil.getConnection();
+    final Table table = conn.getTable(tn);
+    try {
+      List<Put> updates = new ArrayList<>();
+      long bytesToWrite = sizeInBytes;
+      long rowKeyId = 0L;
+      final StringBuilder sb = new StringBuilder();
+      final Random r = new Random();
+      while (bytesToWrite > 0L) {
+        sb.setLength(0);
+        sb.append(Long.toString(rowKeyId));
+        // Use the reverse counter as the rowKey to get even spread across all 
regions
+        Put p = new Put(Bytes.toBytes(sb.reverse().toString()));
+        byte[] value = new byte[SIZE_PER_VALUE];
+        r.nextBytes(value);
+        p.addColumn(Bytes.toBytes(F1), Bytes.toBytes("q1"), value);
+        updates.add(p);
+
+        // Batch ~13KB worth of updates
+        if (updates.size() > 50) {
+          table.put(updates);
+          updates.clear();
+        }
+
+        // Just count the value size, ignore the size of rowkey + column
+        bytesToWrite -= SIZE_PER_VALUE;
+        rowKeyId++;
+      }
+
+      // Write the final batch
+      if (!updates.isEmpty()) {
+        table.put(updates);
+      }
+
+      LOG.debug("Data was written to HBase");
+      // Push the data to disk.
+      testUtil.getAdmin().flush(tn);
+      LOG.debug("Data flushed to disk");
+    } finally {
+      table.close();
+    }
+  }
+
+  Multimap<TableName, QuotaSettings> createTablesWithSpaceQuotas() throws 
Exception {
+    final Admin admin = testUtil.getAdmin();
+    final Multimap<TableName, QuotaSettings> tablesWithQuotas = 
HashMultimap.create();
+
+    final TableName tn1 = createTable();
+    final TableName tn2 = createTable();
+
+    NamespaceDescriptor nd = NamespaceDescriptor.create("ns" + 
counter.getAndIncrement()).build();
+    admin.createNamespace(nd);
+    final TableName tn3 = createTableInNamespace(nd);
+    final TableName tn4 = createTableInNamespace(nd);
+    final TableName tn5 = createTableInNamespace(nd);
+
+    final long sizeLimit1 = 1024L * 1024L * 1024L * 1024L * 5L; // 5TB
+    final SpaceViolationPolicy violationPolicy1 = 
SpaceViolationPolicy.NO_WRITES;
+    QuotaSettings qs1 = QuotaSettingsFactory.limitTableSpace(tn1, sizeLimit1, 
violationPolicy1);
+    tablesWithQuotas.put(tn1, qs1);
+    admin.setQuota(qs1);
+
+    final long sizeLimit2 = 1024L * 1024L * 1024L * 200L; // 200GB
+    final SpaceViolationPolicy violationPolicy2 = 
SpaceViolationPolicy.NO_WRITES_COMPACTIONS;
+    QuotaSettings qs2 = QuotaSettingsFactory.limitTableSpace(tn2, sizeLimit2, 
violationPolicy2);
+    tablesWithQuotas.put(tn2, qs2);
+    admin.setQuota(qs2);
+
+    final long sizeLimit3 = 1024L * 1024L * 1024L * 1024L * 100L; // 100TB
+    final SpaceViolationPolicy violationPolicy3 = 
SpaceViolationPolicy.NO_INSERTS;
+    QuotaSettings qs3 = QuotaSettingsFactory.limitNamespaceSpace(
+        nd.getName(), sizeLimit3, violationPolicy3);
+    tablesWithQuotas.put(tn3, qs3);
+    tablesWithQuotas.put(tn4, qs3);
+    tablesWithQuotas.put(tn5, qs3);
+    admin.setQuota(qs3);
+
+    final long sizeLimit4 = 1024L * 1024L * 1024L * 5L; // 5GB
+    final SpaceViolationPolicy violationPolicy4 = 
SpaceViolationPolicy.NO_INSERTS;
+    QuotaSettings qs4 = QuotaSettingsFactory.limitTableSpace(tn5, sizeLimit4, 
violationPolicy4);
+    // Override the ns quota for tn5, import edge-case to catch table quota 
taking
+    // precedence over ns quota.
+    tablesWithQuotas.put(tn5, qs4);
+    admin.setQuota(qs4);
+
+    return tablesWithQuotas;
+  }
+
+  TableName createTable() throws Exception {
+    return createTableWithRegions(1);
+  }
+
+  TableName createTableWithRegions(int numRegions) throws Exception {
+    return 
createTableWithRegions(NamespaceDescriptor.DEFAULT_NAMESPACE_NAME_STR, 
numRegions);
+  }
+
+  TableName createTableWithRegions(String namespace, int numRegions) throws 
Exception {
+    final Admin admin = testUtil.getAdmin();
+    final TableName tn = TableName.valueOf(
+        namespace, testName.getMethodName() + counter.getAndIncrement());
+
+    // Delete the old table
+    if (admin.tableExists(tn)) {
+      admin.disableTable(tn);
+      admin.deleteTable(tn);
+    }
+
+    // Create the table
+    HTableDescriptor tableDesc = new HTableDescriptor(tn);
+    tableDesc.addFamily(new HColumnDescriptor(F1));
+    if (numRegions == 1) {
+      admin.createTable(tableDesc);
+    } else {
+      admin.createTable(tableDesc, Bytes.toBytes("0"), Bytes.toBytes("9"), 
numRegions);
+    }
+    return tn;
+  }
+
+  TableName createTableInNamespace(NamespaceDescriptor nd) throws Exception {
+    final Admin admin = testUtil.getAdmin();
+    final TableName tn = TableName.valueOf(nd.getName(),
+        testName.getMethodName() + counter.getAndIncrement());
+
+    // Delete the old table
+    if (admin.tableExists(tn)) {
+      admin.disableTable(tn);
+      admin.deleteTable(tn);
+    }
+
+    // Create the table
+    HTableDescriptor tableDesc = new HTableDescriptor(tn);
+    tableDesc.addFamily(new HColumnDescriptor(F1));
+
+    admin.createTable(tableDesc);
+    return tn;
+  }
+
+  void partitionTablesByQuotaTarget(Multimap<TableName,QuotaSettings> quotas,
+      Set<TableName> tablesWithTableQuota, Set<TableName> 
tablesWithNamespaceQuota) {
+    // Partition the tables with quotas by table and ns quota
+    for (Entry<TableName, QuotaSettings> entry : quotas.entries()) {
+      SpaceLimitSettings settings = (SpaceLimitSettings) entry.getValue();
+      TableName tn = entry.getKey();
+      if (null != settings.getTableName()) {
+        tablesWithTableQuota.add(tn);
+      }
+      if (null != settings.getNamespace()) {
+        tablesWithNamespaceQuota.add(tn);
+      }
+
+      if (null == settings.getTableName() && null == settings.getNamespace()) {
+        fail("Unexpected table name with null tableName and namespace: " + tn);
+      }
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/SpaceQuotaSnapshotNotifierForTest.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/SpaceQuotaSnapshotNotifierForTest.java
 
b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/SpaceQuotaSnapshotNotifierForTest.java
new file mode 100644
index 0000000..0986e8c
--- /dev/null
+++ 
b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/SpaceQuotaSnapshotNotifierForTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.HashMap;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.client.Connection;
+
+/**
+ * A SpaceQuotaSnapshotNotifier implementation for testing.
+ */
+@InterfaceAudience.Private
+public class SpaceQuotaSnapshotNotifierForTest implements 
SpaceQuotaSnapshotNotifier {
+  private static final Log LOG = 
LogFactory.getLog(SpaceQuotaSnapshotNotifierForTest.class);
+
+  private final Map<TableName,SpaceQuotaSnapshot> tableQuotaSnapshots = new 
HashMap<>();
+
+  @Override
+  public void initialize(Connection conn) {}
+
+  @Override
+  public synchronized void transitionTable(TableName tableName, 
SpaceQuotaSnapshot snapshot) {
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("Persisting " + tableName + "=>" + snapshot);
+    }
+    tableQuotaSnapshots.put(tableName, snapshot);
+  }
+
+  public synchronized Map<TableName,SpaceQuotaSnapshot> copySnapshots() {
+    return new HashMap<>(this.tableQuotaSnapshots);
+  }
+
+  public synchronized void clearSnapshots() {
+    this.tableQuotaSnapshots.clear();
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestActivePolicyEnforcement.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestActivePolicyEnforcement.java
 
b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestActivePolicyEnforcement.java
new file mode 100644
index 0000000..80363e8
--- /dev/null
+++ 
b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestActivePolicyEnforcement.java
@@ -0,0 +1,74 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.hbase.TableName;
+import 
org.apache.hadoop.hbase.quotas.policies.NoWritesViolationPolicyEnforcement;
+import org.apache.hadoop.hbase.regionserver.RegionServerServices;
+import 
org.apache.hadoop.hbase.quotas.policies.BulkLoadVerifyingViolationPolicyEnforcement;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+/**
+ * Test class for {@link ActivePolicyEnforcement}.
+ */
+@Category(SmallTests.class)
+public class TestActivePolicyEnforcement {
+
+  @Test
+  public void testGetter() {
+    final TableName tableName = TableName.valueOf("table");
+    Map<TableName, SpaceViolationPolicyEnforcement> map = new HashMap<>();
+    map.put(tableName, new NoWritesViolationPolicyEnforcement());
+    ActivePolicyEnforcement ape = new ActivePolicyEnforcement(map, 
Collections.emptyMap(), null);
+    assertEquals(map.get(tableName), ape.getPolicyEnforcement(tableName));
+  }
+
+  @Test
+  public void testNoPolicyReturnsNoopEnforcement() {
+    ActivePolicyEnforcement ape = new ActivePolicyEnforcement(
+        new HashMap<>(), Collections.emptyMap(), 
mock(RegionServerServices.class));
+    SpaceViolationPolicyEnforcement enforcement = ape.getPolicyEnforcement(
+        TableName.valueOf("nonexistent"));
+    assertNotNull(enforcement);
+    assertTrue(
+        "Expected an instance of NoopViolationPolicyEnforcement",
+        enforcement instanceof BulkLoadVerifyingViolationPolicyEnforcement);
+  }
+
+  @Test
+  public void testNoBulkLoadChecksOnNoSnapshot() {
+    ActivePolicyEnforcement ape = new ActivePolicyEnforcement(
+        new HashMap<TableName, SpaceViolationPolicyEnforcement>(),
+        Collections.<TableName,SpaceQuotaSnapshot> emptyMap(),
+        mock(RegionServerServices.class));
+    SpaceViolationPolicyEnforcement enforcement = ape.getPolicyEnforcement(
+        TableName.valueOf("nonexistent"));
+    assertFalse("Should not check bulkloads", 
enforcement.shouldCheckBulkLoads());
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestFileSystemUtilizationChore.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestFileSystemUtilizationChore.java
 
b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestFileSystemUtilizationChore.java
index ad98720..18e47af 100644
--- 
a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestFileSystemUtilizationChore.java
+++ 
b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestFileSystemUtilizationChore.java
@@ -144,6 +144,7 @@ public class TestFileSystemUtilizationChore {
     assertEquals(timeUnit, chore.getTimeUnit());
   }
 
+  @SuppressWarnings("unchecked")
   @Test
   public void testProcessingLeftoverRegions() {
     final Configuration conf = getDefaultHBaseConfiguration();
@@ -176,6 +177,7 @@ public class TestFileSystemUtilizationChore {
     chore.chore();
   }
 
+  @SuppressWarnings("unchecked")
   @Test
   public void testProcessingNowOfflineLeftoversAreIgnored() {
     final Configuration conf = getDefaultHBaseConfiguration();
@@ -185,7 +187,6 @@ public class TestFileSystemUtilizationChore {
     final List<Long> leftover1Sizes = Arrays.asList(1024L, 4096L);
     final long leftover1Sum = sum(leftover1Sizes);
     final List<Long> leftover2Sizes = Arrays.asList(2048L);
-    final long leftover2Sum = sum(leftover2Sizes);
 
     final Region lr1 = mockRegionWithSize(leftover1Sizes);
     final Region lr2 = mockRegionWithSize(leftover2Sizes);

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestNamespaceQuotaViolationStore.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestNamespaceQuotaViolationStore.java
 
b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestNamespaceQuotaViolationStore.java
index 8182513..4a7258f 100644
--- 
a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestNamespaceQuotaViolationStore.java
+++ 
b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestNamespaceQuotaViolationStore.java
@@ -31,7 +31,6 @@ import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.NamespaceDescriptor;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Connection;
-import org.apache.hadoop.hbase.quotas.QuotaViolationStore.ViolationState;
 import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.Quotas;
@@ -45,7 +44,7 @@ import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
 /**
- * Test class for {@link NamespaceQuotaViolationStore}.
+ * Test class for {@link NamespaceQuotaSnapshotStore}.
  */
 @Category(SmallTests.class)
 public class TestNamespaceQuotaViolationStore {
@@ -54,19 +53,19 @@ public class TestNamespaceQuotaViolationStore {
   private Connection conn;
   private QuotaObserverChore chore;
   private Map<HRegionInfo, Long> regionReports;
-  private NamespaceQuotaViolationStore store;
+  private NamespaceQuotaSnapshotStore store;
 
   @Before
   public void setup() {
     conn = mock(Connection.class);
     chore = mock(QuotaObserverChore.class);
     regionReports = new HashMap<>();
-    store = new NamespaceQuotaViolationStore(conn, chore, regionReports);
+    store = new NamespaceQuotaSnapshotStore(conn, chore, regionReports);
   }
 
   @Test
   public void testGetSpaceQuota() throws Exception {
-    NamespaceQuotaViolationStore mockStore = 
mock(NamespaceQuotaViolationStore.class);
+    NamespaceQuotaSnapshotStore mockStore = 
mock(NamespaceQuotaSnapshotStore.class);
     when(mockStore.getSpaceQuota(any(String.class))).thenCallRealMethod();
 
     Quotas quotaWithSpace = Quotas.newBuilder().setSpace(
@@ -113,17 +112,18 @@ public class TestNamespaceQuotaViolationStore {
     regionReports.put(new HRegionInfo(tn1, Bytes.toBytes(1), 
Bytes.toBytes(2)), 1024L * 256L);
 
     // Below the quota
-    assertEquals(ViolationState.IN_OBSERVANCE, store.getTargetState(NS, 
quota));
+    assertEquals(false, store.getTargetState(NS, 
quota).getQuotaStatus().isInViolation());
 
     regionReports.put(new HRegionInfo(tn2, Bytes.toBytes(2), 
Bytes.toBytes(3)), 1024L * 256L);
 
     // Equal to the quota is still in observance
-    assertEquals(ViolationState.IN_OBSERVANCE, store.getTargetState(NS, 
quota));
+    assertEquals(false, store.getTargetState(NS, 
quota).getQuotaStatus().isInViolation());
 
     regionReports.put(new HRegionInfo(tn2, Bytes.toBytes(3), 
Bytes.toBytes(4)), 1024L);
 
     // Exceeds the quota, should be in violation
-    assertEquals(ViolationState.IN_VIOLATION, store.getTargetState(NS, quota));
+    assertEquals(true, store.getTargetState(NS, 
quota).getQuotaStatus().isInViolation());
+    assertEquals(SpaceViolationPolicy.DISABLE, store.getTargetState(NS, 
quota).getQuotaStatus().getPolicy());
   }
 
   @Test

http://git-wip-us.apache.org/repos/asf/hbase/blob/b99e9cf9/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaObserverChore.java
----------------------------------------------------------------------
diff --git 
a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaObserverChore.java
 
b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaObserverChore.java
index db549e4..da294c6 100644
--- 
a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaObserverChore.java
+++ 
b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaObserverChore.java
@@ -17,8 +17,6 @@
 package org.apache.hadoop.hbase.quotas;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
-import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -28,8 +26,6 @@ import java.util.Map;
 import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Connection;
-import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
-import 
org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.SpaceQuota;
 import org.apache.hadoop.hbase.testclassification.SmallTests;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.junit.Before;
@@ -50,8 +46,6 @@ public class TestQuotaObserverChore {
   public void setup() throws Exception {
     conn = mock(Connection.class);
     chore = mock(QuotaObserverChore.class);
-    // Set up some rules to call the real method on the mock.
-    when(chore.getViolationPolicy(any(SpaceQuota.class))).thenCallRealMethod();
   }
 
   @Test
@@ -76,31 +70,11 @@ public class TestQuotaObserverChore {
       regionReports.put(new HRegionInfo(tn3, Bytes.toBytes(i), Bytes.toBytes(i 
+ 1)), 0L);
     }
 
-    TableQuotaViolationStore store = new TableQuotaViolationStore(conn, chore, 
regionReports);
-    when(chore.getTableViolationStore()).thenReturn(store);
+    TableQuotaSnapshotStore store = new TableQuotaSnapshotStore(conn, chore, 
regionReports);
+    when(chore.getTableSnapshotStore()).thenReturn(store);
 
     assertEquals(numTable1Regions, Iterables.size(store.filterBySubject(tn1)));
     assertEquals(numTable2Regions, Iterables.size(store.filterBySubject(tn2)));
     assertEquals(numTable3Regions, Iterables.size(store.filterBySubject(tn3)));
   }
-
-  @Test
-  public void testExtractViolationPolicy() {
-    for (SpaceViolationPolicy policy : SpaceViolationPolicy.values()) {
-      SpaceQuota spaceQuota = SpaceQuota.newBuilder()
-          .setSoftLimit(1024L)
-          .setViolationPolicy(ProtobufUtil.toProtoViolationPolicy(policy))
-          .build();
-      assertEquals(policy, chore.getViolationPolicy(spaceQuota));
-    }
-    SpaceQuota malformedQuota = SpaceQuota.newBuilder()
-        .setSoftLimit(1024L)
-        .build();
-    try {
-      chore.getViolationPolicy(malformedQuota);
-      fail("Should have thrown an IllegalArgumentException.");
-    } catch (IllegalArgumentException e) {
-      // Pass
-    }
-  }
 }
\ No newline at end of file

Reply via email to