From 093235e0124d5bbbaab75feb510e6cf4169d0f63 Mon Sep 17 00:00:00 2001 From: meiyi Date: Fri, 25 Jan 2019 19:31:21 +0800 Subject: [PATCH] HBASE-21783 Support allow exceed user/table/ns rpc throttle quota if region server has available quota --- .../hadoop/hbase/quotas/QuotaSettingsFactory.java | 23 +++++++ .../hadoop/hbase/quotas/ThrottleSettings.java | 11 ++++ .../hadoop/hbase/shaded/protobuf/ProtobufUtil.java | 16 +++++ .../src/main/protobuf/Quota.proto | 1 + .../hbase/quotas/AllowExceedOperationQuota.java | 74 ++++++++++++++++++++++ .../hadoop/hbase/quotas/DefaultOperationQuota.java | 30 +++++---- .../hbase/quotas/GlobalQuotaSettingsImpl.java | 3 + .../hadoop/hbase/quotas/NoopQuotaLimiter.java | 5 ++ .../apache/hadoop/hbase/quotas/QuotaLimiter.java | 3 + .../apache/hadoop/hbase/quotas/RateLimiter.java | 23 ++++++- .../hbase/quotas/RegionServerRpcQuotaManager.java | 9 ++- .../hadoop/hbase/quotas/TimeBasedLimiter.java | 14 +++- .../apache/hadoop/hbase/quotas/TestQuotaAdmin.java | 21 +++--- .../hadoop/hbase/quotas/TestQuotaThrottle.java | 18 ++++++ hbase-shell/src/main/ruby/hbase/quotas.rb | 6 +- hbase-shell/src/main/ruby/hbase_constants.rb | 1 + hbase-shell/src/test/ruby/hbase/quotas_test.rb | 7 +- 17 files changed, 235 insertions(+), 30 deletions(-) create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/AllowExceedOperationQuota.java diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/quotas/QuotaSettingsFactory.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/quotas/QuotaSettingsFactory.java index 48f7f65..e4b04fe 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/quotas/QuotaSettingsFactory.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/quotas/QuotaSettingsFactory.java @@ -331,6 +331,29 @@ public class QuotaSettingsFactory { } /** + * Throttle the specified region server. + * + * @param regionServer the region server to throttle + * @param type the type of throttling + * @param limit the allowed number of request/data per timeUnit + * @param timeUnit the limit time unit + * @param allowExceed if allow exceed quota + * @return the quota settings + */ + public static QuotaSettings throttleRegionServer(final String regionServer, + final ThrottleType type, final long limit, final TimeUnit timeUnit, boolean allowExceed) { + QuotaProtos.ThrottleRequest.Builder builder = QuotaProtos.ThrottleRequest.newBuilder(); + if (type != null) { + builder.setType(ProtobufUtil.toProtoThrottleType(type)); + } + if (timeUnit != null) { + builder.setTimedQuota( + ProtobufUtil.toTimedQuota(limit, timeUnit, QuotaScope.MACHINE, allowExceed)); + } + return new ThrottleSettings(null, null, null, regionServer, builder.build()); + } + + /** * Remove the throttling for the specified region server. * * @param regionServer the region Server diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/quotas/ThrottleSettings.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/quotas/ThrottleSettings.java index c616620..7bae8c4 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/quotas/ThrottleSettings.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/quotas/ThrottleSettings.java @@ -48,6 +48,14 @@ class ThrottleSettings extends QuotaSettings { public long getSoftLimit() { return proto.hasTimedQuota() ? proto.getTimedQuota().getSoftLimit() : -1; } + + public boolean allowExceed() { + if (proto.hasTimedQuota()) { + return proto.getTimedQuota().hasAllowExceed() ? proto.getTimedQuota().getAllowExceed() + : false; + } + return false; + } /** * Returns a copy of the internal state of this @@ -111,6 +119,9 @@ class ThrottleSettings extends QuotaSettings { builder.append(", SCOPE => "); builder.append(timedQuota.getScope().toString()); } + if (timedQuota.hasAllowExceed() && timedQuota.getAllowExceed()) { + builder.append(", ALLOW_EXCEED => true"); + } } else { builder.append(", LIMIT => NONE"); } diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java index fea81f1..d05676d 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java @@ -2548,6 +2548,22 @@ public final class ProtobufUtil { } /** + * Build a protocol buffer TimedQuota, only used for region server throttle quota because + * allowExceed param may be true + * @param limit the allowed number of request/data per timeUnit + * @param timeUnit the limit time unit + * @param scope the quota scope + * @param allowExceed True if allow exceed quota + * @return the protocol buffer TimedQuota + */ + public static QuotaProtos.TimedQuota toTimedQuota(final long limit, final TimeUnit timeUnit, + final QuotaScope scope, boolean allowExceed) { + return QuotaProtos.TimedQuota.newBuilder().setSoftLimit(limit) + .setTimeUnit(toProtoTimeUnit(timeUnit)).setScope(toProtoQuotaScope(scope)) + .setAllowExceed(allowExceed).build(); + } + + /** * Builds a protocol buffer SpaceQuota. * * @param limit The maximum space usage for the quota in bytes. diff --git a/hbase-protocol-shaded/src/main/protobuf/Quota.proto b/hbase-protocol-shaded/src/main/protobuf/Quota.proto index 5b00d74..3e21697 100644 --- a/hbase-protocol-shaded/src/main/protobuf/Quota.proto +++ b/hbase-protocol-shaded/src/main/protobuf/Quota.proto @@ -37,6 +37,7 @@ message TimedQuota { optional uint64 soft_limit = 2; optional float share = 3; optional QuotaScope scope = 4 [default = MACHINE]; + optional bool allow_exceed = 5 [default = false]; } enum ThrottleType { diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/AllowExceedOperationQuota.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/AllowExceedOperationQuota.java new file mode 100644 index 0000000..18e8350 --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/AllowExceedOperationQuota.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 org.apache.hadoop.conf.Configuration; +import org.apache.yetus.audience.InterfaceAudience; +import org.apache.yetus.audience.InterfaceStability; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +@InterfaceAudience.Private +@InterfaceStability.Evolving +public class AllowExceedOperationQuota extends DefaultOperationQuota { + private static final Logger LOG = LoggerFactory.getLogger(AllowExceedOperationQuota.class); + private QuotaLimiter regionServerLimiter; + + public AllowExceedOperationQuota(final Configuration conf, QuotaLimiter regionServerLimiter, + final QuotaLimiter... limiters) { + super(conf, limiters); + this.regionServerLimiter = regionServerLimiter; + } + + @Override + public void checkQuota(int numWrites, int numReads, int numScans) throws RpcThrottlingException { + boolean exceed = false; + try { + super.checkQuota(numWrites, numReads, numScans); + } catch (RpcThrottlingException e) { + exceed = true; + if (LOG.isDebugEnabled()) { + LOG.debug("Read/Write requests num exceeds quota: writes:{} reads:{} scan:{}, " + + "try use region server quota", + numWrites, numReads, numScans); + } + } + if (regionServerLimiter.isBypass()) return; + + regionServerLimiter.checkQuota(numWrites, writeConsumed, numReads + numScans, readConsumed, + writeCapacityUnitConsumed, readCapacityUnitConsumed); + readAvailable = Math.max(readAvailable, regionServerLimiter.getReadAvailable()); + writeAvailable = Math.max(writeAvailable, regionServerLimiter.getWriteAvailable()); + regionServerLimiter.grabQuota(numWrites, writeConsumed, numReads + numScans, readConsumed, + writeCapacityUnitConsumed, writeCapacityUnitConsumed); + if (exceed) { + for (final QuotaLimiter limiter : limiters) { + limiter.grabQuota(numWrites, writeConsumed, numReads + numScans, readConsumed, + writeCapacityUnitConsumed, writeCapacityUnitConsumed); + } + } + } + + @Override + public void close() { + super.close(); + if (writeDiff != 0) regionServerLimiter.consumeWrite(writeDiff, writeCapacityUnitDiff); + if (readDiff != 0) regionServerLimiter.consumeRead(readDiff, readCapacityUnitDiff); + } +} diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/DefaultOperationQuota.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/DefaultOperationQuota.java index f9b3ca5..9e27d8e 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/DefaultOperationQuota.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/DefaultOperationQuota.java @@ -34,17 +34,25 @@ import org.apache.hadoop.hbase.client.Result; public class DefaultOperationQuota implements OperationQuota { private static final Logger LOG = LoggerFactory.getLogger(DefaultOperationQuota.class); - private final List limiters; + protected final List limiters; private final long writeCapacityUnit; private final long readCapacityUnit; - private long writeAvailable = 0; - private long readAvailable = 0; - private long writeConsumed = 0; - private long readConsumed = 0; - private long writeCapacityUnitConsumed = 0; - private long readCapacityUnitConsumed = 0; + // the available read/write quota size in bytes + protected long writeAvailable = 0; + protected long readAvailable = 0; + // estimated quota + protected long writeConsumed = 0; + protected long readConsumed = 0; + protected long writeCapacityUnitConsumed = 0; + protected long readCapacityUnitConsumed = 0; + // real consumed quota private final long[] operationSize; + // difference between estimated quota and real consumed quota + protected long writeDiff = 0; + protected long readDiff = 0; + protected long writeCapacityUnitDiff = 0; + protected long readCapacityUnitDiff = 0; public DefaultOperationQuota(final Configuration conf, final QuotaLimiter... limiters) { this(conf, Arrays.asList(limiters)); @@ -96,12 +104,12 @@ public class DefaultOperationQuota implements OperationQuota { @Override public void close() { // Adjust the quota consumed for the specified operation - long writeDiff = operationSize[OperationType.MUTATE.ordinal()] - writeConsumed; - long readDiff = operationSize[OperationType.GET.ordinal()] + writeDiff = operationSize[OperationType.MUTATE.ordinal()] - writeConsumed; + readDiff = operationSize[OperationType.GET.ordinal()] + operationSize[OperationType.SCAN.ordinal()] - readConsumed; - long writeCapacityUnitDiff = calculateWriteCapacityUnitDiff( + writeCapacityUnitDiff = calculateWriteCapacityUnitDiff( operationSize[OperationType.MUTATE.ordinal()], writeConsumed); - long readCapacityUnitDiff = calculateReadCapacityUnitDiff( + readCapacityUnitDiff = calculateReadCapacityUnitDiff( operationSize[OperationType.GET.ordinal()] + operationSize[OperationType.SCAN.ordinal()], readConsumed); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/GlobalQuotaSettingsImpl.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/GlobalQuotaSettingsImpl.java index c7df789..18d89a2 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/GlobalQuotaSettingsImpl.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/GlobalQuotaSettingsImpl.java @@ -261,6 +261,9 @@ public class GlobalQuotaSettingsImpl extends GlobalQuotaSettings { builder.append(", SCOPE => "); builder.append(timedQuota.getScope().toString()); } + if (timedQuota.hasAllowExceed() && timedQuota.getAllowExceed()) { + builder.append(", ALLOW_EXCEED => true"); + } } builder.append( "} } "); } else { diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/NoopQuotaLimiter.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/NoopQuotaLimiter.java index 71dd3c7..38508fa 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/NoopQuotaLimiter.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/NoopQuotaLimiter.java @@ -79,4 +79,9 @@ class NoopQuotaLimiter implements QuotaLimiter { public static QuotaLimiter get() { return instance; } + + @Override + public boolean isAllowExceed() { + return false; + } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/QuotaLimiter.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/QuotaLimiter.java index 9260ec2..244affd 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/QuotaLimiter.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/QuotaLimiter.java @@ -78,4 +78,7 @@ public interface QuotaLimiter { /** @return the number of bytes available to write to avoid exceeding the quota */ long getWriteAvailable(); + + /** @return True if allow exceed quota */ + boolean isAllowExceed(); } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/RateLimiter.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/RateLimiter.java index 852d8a6..3402b6a 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/RateLimiter.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/RateLimiter.java @@ -53,6 +53,7 @@ public abstract class RateLimiter { private long tunit = 1000; // Timeunit factor for translating to ms. private long limit = Long.MAX_VALUE; // The max value available resource units can be refilled to. private long avail = Long.MAX_VALUE; // Currently available resource units + private boolean allowExceed = false; /** * Refill the available units w.r.t the elapsed time. @@ -70,13 +71,23 @@ public abstract class RateLimiter { */ abstract long getWaitInterval(long limit, long available, long amount); - /** * Set the RateLimiter max available resources and refill period. * @param limit The max value available resource units can be refilled to. * @param timeUnit Timeunit factor for translating to ms. */ public synchronized void set(final long limit, final TimeUnit timeUnit) { + set(limit, timeUnit, false); + } + + /** + * Set the RateLimiter max available resources and refill period. + * @param limit The max value available resource units can be refilled to. + * @param timeUnit Timeunit factor for translating to ms. + * @param allowExceed If allow exceed quota, only used by region server quota + */ + public synchronized void set(final long limit, final TimeUnit timeUnit, + final boolean allowExceed) { switch (timeUnit) { case MILLISECONDS: tunit = 1; @@ -98,6 +109,7 @@ public abstract class RateLimiter { } this.limit = limit; this.avail = limit; + this.allowExceed = allowExceed; } @Override @@ -106,8 +118,9 @@ public abstract class RateLimiter { if (getLimit() == Long.MAX_VALUE) { return rateLimiter + "(Bypass)"; } - return rateLimiter + "(avail=" + getAvailable() + " limit=" + getLimit() + - " tunit=" + getTimeUnitInMillis() + ")"; + return new StringBuilder().append(rateLimiter).append("(avail=").append(getAvailable()) + .append(" limit=").append(getLimit()).append(" tunit=").append(getTimeUnitInMillis()) + .append(" allowExceed=").append(isAllowExceed()).append(")").toString(); } /** @@ -235,4 +248,8 @@ public abstract class RateLimiter { @VisibleForTesting public abstract long getNextRefillTime(); + + public boolean isAllowExceed() { + return allowExceed; + } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/RegionServerRpcQuotaManager.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/RegionServerRpcQuotaManager.java index 99ff516..84ea066 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/RegionServerRpcQuotaManager.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/RegionServerRpcQuotaManager.java @@ -144,8 +144,13 @@ public class RegionServerRpcQuotaManager { + rsLimiter); } if (!useNoop) { - return new DefaultOperationQuota(this.rsServices.getConfiguration(), userLimiter, - tableLimiter, nsLimiter, rsLimiter); + if (!rsLimiter.isAllowExceed()) { + return new DefaultOperationQuota(this.rsServices.getConfiguration(), userLimiter, + tableLimiter, nsLimiter, rsLimiter); + } else { + return new AllowExceedOperationQuota(this.rsServices.getConfiguration(), rsLimiter, + userLimiter, tableLimiter, nsLimiter); + } } } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TimeBasedLimiter.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TimeBasedLimiter.java index 6b5349f..59c2dc9 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TimeBasedLimiter.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TimeBasedLimiter.java @@ -133,7 +133,8 @@ public class TimeBasedLimiter implements QuotaLimiter { } private static void setFromTimedQuota(final RateLimiter limiter, final TimedQuota timedQuota) { - limiter.set(timedQuota.getSoftLimit(), ProtobufUtil.toTimeUnit(timedQuota.getTimeUnit())); + limiter.set(timedQuota.getSoftLimit(), ProtobufUtil.toTimeUnit(timedQuota.getTimeUnit()), + timedQuota.hasAllowExceed() ? timedQuota.getAllowExceed() : false); } @Override @@ -272,4 +273,15 @@ public class TimeBasedLimiter implements QuotaLimiter { builder.append(')'); return builder.toString(); } + + public boolean isAllowExceed() { + if (!isBypass()) { + return reqsLimiter.isAllowExceed() | reqSizeLimiter.isAllowExceed() + | reqCapacityUnitLimiter.isAllowExceed() | readReqsLimiter.isAllowExceed() + | readSizeLimiter.isAllowExceed() | writeReqsLimiter.isAllowExceed() + | writeSizeLimiter.isAllowExceed() | readCapacityUnitLimiter.isAllowExceed() + | writeCapacityUnitLimiter.isAllowExceed(); + } + return false; + } } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaAdmin.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaAdmin.java index 88cd127..1f724b6 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaAdmin.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaAdmin.java @@ -471,7 +471,7 @@ public class TestQuotaAdmin { admin.setQuota(settings); // Verify the Quota in the table - verifyRecordPresentInQuotaTable(throttleType, 2L, TimeUnit.HOURS); + verifyRecordPresentInQuotaTable(throttleType, 2L, TimeUnit.HOURS, false); // Verify we can retrieve it via the QuotaRetriever API verifyFetchableViaAPI(admin, throttleType, 2L, TimeUnit.HOURS); @@ -496,7 +496,7 @@ public class TestQuotaAdmin { admin.setQuota(settings); // Verify the Quota in the table - verifyRecordPresentInQuotaTable(ThrottleType.REQUEST_SIZE, 2L, TimeUnit.HOURS); + verifyRecordPresentInQuotaTable(ThrottleType.REQUEST_SIZE, 2L, TimeUnit.HOURS, false); // Verify we can retrieve it via the QuotaRetriever API verifyFetchableViaAPI(admin, ThrottleType.REQUEST_SIZE, 2L, TimeUnit.HOURS); @@ -507,7 +507,7 @@ public class TestQuotaAdmin { admin.setQuota(newSettings); // Verify the new Quota in the table - verifyRecordPresentInQuotaTable(ThrottleType.REQUEST_SIZE, 3L, TimeUnit.DAYS); + verifyRecordPresentInQuotaTable(ThrottleType.REQUEST_SIZE, 3L, TimeUnit.DAYS, false); // Verify we can retrieve the new quota via the QuotaRetriever API verifyFetchableViaAPI(admin, ThrottleType.REQUEST_SIZE, 3L, TimeUnit.DAYS); @@ -534,13 +534,13 @@ public class TestQuotaAdmin { ThrottleType.REQUEST_NUMBER, 10, TimeUnit.MINUTES)); assertNumResults(1, rsFilter); // Verify the Quota in the table - verifyRecordPresentInQuotaTable(ThrottleType.REQUEST_NUMBER, 10, TimeUnit.MINUTES); + verifyRecordPresentInQuotaTable(ThrottleType.REQUEST_NUMBER, 10, TimeUnit.MINUTES, false); admin.setQuota(QuotaSettingsFactory.throttleRegionServer(regionServer, - ThrottleType.REQUEST_NUMBER, 20, TimeUnit.MINUTES)); + ThrottleType.REQUEST_NUMBER, 20, TimeUnit.MINUTES, true)); assertNumResults(1, rsFilter); // Verify the Quota in the table - verifyRecordPresentInQuotaTable(ThrottleType.REQUEST_NUMBER, 20, TimeUnit.MINUTES); + verifyRecordPresentInQuotaTable(ThrottleType.REQUEST_NUMBER, 20, TimeUnit.MINUTES, true); admin.setQuota(QuotaSettingsFactory.throttleRegionServer(regionServer, ThrottleType.READ_NUMBER, 30, TimeUnit.SECONDS)); @@ -555,9 +555,11 @@ public class TestQuotaAdmin { if (throttleSettings.getThrottleType() == ThrottleType.REQUEST_NUMBER) { assertEquals(20, throttleSettings.getSoftLimit()); assertEquals(TimeUnit.MINUTES, throttleSettings.getTimeUnit()); + assertEquals(true, throttleSettings.allowExceed()); } else if (throttleSettings.getThrottleType() == ThrottleType.READ_NUMBER) { assertEquals(30, throttleSettings.getSoftLimit()); assertEquals(TimeUnit.SECONDS, throttleSettings.getTimeUnit()); + assertEquals(false, throttleSettings.allowExceed()); } } } finally { @@ -607,7 +609,7 @@ public class TestQuotaAdmin { rs.getRegionServer().getRegionServerRpcQuotaManager().isRpcThrottleEnabled())); } - private void verifyRecordPresentInQuotaTable(ThrottleType type, long limit, TimeUnit tu) + private void verifyRecordPresentInQuotaTable(ThrottleType type, long limit, TimeUnit tu, boolean allowExceed) throws Exception { // Verify the RPC Quotas in the table try (Table quotaTable = TEST_UTIL.getConnection().getTable(QuotaTableUtil.QUOTA_TABLE_NAME); @@ -615,7 +617,7 @@ public class TestQuotaAdmin { Result r = Iterables.getOnlyElement(scanner); CellScanner cells = r.cellScanner(); assertTrue("Expected to find a cell", cells.advance()); - assertRPCQuota(type, limit, tu, cells.current()); + assertRPCQuota(type, limit, tu, allowExceed, cells.current()); } } @@ -642,7 +644,7 @@ public class TestQuotaAdmin { } } - private void assertRPCQuota(ThrottleType type, long limit, TimeUnit tu, Cell cell) + private void assertRPCQuota(ThrottleType type, long limit, TimeUnit tu, boolean allowExceed, Cell cell) throws Exception { Quotas q = QuotaTableUtil .quotasFromData(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()); @@ -693,6 +695,7 @@ public class TestQuotaAdmin { assertEquals(t.getSoftLimit(), limit); assertEquals(t.getTimeUnit(), ProtobufUtil.toProtoTimeUnit(tu)); + assertEquals(allowExceed, t.getAllowExceed()); } private void assertRPCQuota(ThrottleType type, long limit, TimeUnit tu, diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaThrottle.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaThrottle.java index aee9707..396fa13 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaThrottle.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaThrottle.java @@ -595,6 +595,24 @@ public class TestQuotaThrottle { triggerCacheRefresh(true, false, true, false, true, TABLE_NAMES[0]); } + @Test + public void testRegionServerThrottleAllowExceed() throws Exception { + final Admin admin = TEST_UTIL.getAdmin(); + admin.setQuota(QuotaSettingsFactory.throttleTable(TABLE_NAMES[0], ThrottleType.WRITE_NUMBER, 5, + TimeUnit.MINUTES)); + // allow exceed region server quota + admin.setQuota(QuotaSettingsFactory.throttleRegionServer(QuotaTableUtil.QUOTA_REGION_SERVER_ROW_KEY, + ThrottleType.WRITE_NUMBER, 20, TimeUnit.MINUTES, true)); + triggerCacheRefresh(false, false, true, false, true, TABLE_NAMES[0]); + waitMinuteQuota(); + assertEquals(10, doPuts(10, tables[0])); + + // unthrottle + admin.setQuota(QuotaSettingsFactory.unthrottleRegionServer(QuotaTableUtil.QUOTA_REGION_SERVER_ROW_KEY)); + admin.setQuota(QuotaSettingsFactory.unthrottleTable(TABLE_NAMES[0])); + triggerCacheRefresh(true, false, true, false, true, TABLE_NAMES[0]); + } + private int doPuts(int maxOps, final Table... tables) throws Exception { return doPuts(maxOps, -1, tables); } diff --git a/hbase-shell/src/main/ruby/hbase/quotas.rb b/hbase-shell/src/main/ruby/hbase/quotas.rb index 4023aed..fa6dffd 100644 --- a/hbase-shell/src/main/ruby/hbase/quotas.rb +++ b/hbase-shell/src/main/ruby/hbase/quotas.rb @@ -83,8 +83,12 @@ module Hbase raise(ArgumentError, 'Unexpected arguments: ' + args.inspect) unless args.empty? settings = QuotaSettingsFactory.throttleNamespace(namespace, type, limit, time_unit) elsif args.key?(REGIONSERVER) + args.delete(REGIONSERVER) + allow_exceed = args.fetch(ALLOW_EXCEED, FALSE) + args.delete(ALLOW_EXCEED) + raise(ArgumentError, 'Unexpected arguments: ' + args.inspect) unless args.empty? # TODO: Setting specified region server quota isn't supported currently and using 'all' for all RS - settings = QuotaSettingsFactory.throttleRegionServer('all', type, limit, time_unit) + settings = QuotaSettingsFactory.throttleRegionServer('all', type, limit, time_unit, allow_exceed) else raise 'One of USER, TABLE, NAMESPACE or REGIONSERVER must be specified' end diff --git a/hbase-shell/src/main/ruby/hbase_constants.rb b/hbase-shell/src/main/ruby/hbase_constants.rb index 9871685..447a1e3 100644 --- a/hbase-shell/src/main/ruby/hbase_constants.rb +++ b/hbase-shell/src/main/ruby/hbase_constants.rb @@ -93,6 +93,7 @@ module HBaseConstants FORMATTER_CLASS = 'FORMATTER_CLASS'.freeze POLICY = 'POLICY'.freeze REGIONSERVER = 'REGIONSERVER'.freeze + ALLOW_EXCEED = 'ALLOW_EXCEED'.freeze # Load constants from hbase java API def self.promote_constants(constants) diff --git a/hbase-shell/src/test/ruby/hbase/quotas_test.rb b/hbase-shell/src/test/ruby/hbase/quotas_test.rb index 1bcc07b..37020a8 100644 --- a/hbase-shell/src/test/ruby/hbase/quotas_test.rb +++ b/hbase-shell/src/test/ruby/hbase/quotas_test.rb @@ -260,13 +260,14 @@ module Hbase assert(output.include?('THROTTLE_TYPE => REQUEST_CAPACITY_UNIT')) assert(output.include?('LIMIT => 1CU/sec')) - command(:set_quota, TYPE => THROTTLE, REGIONSERVER => 'all', THROTTLE_TYPE => WRITE, LIMIT => '2req/sec') + command(:set_quota, TYPE => THROTTLE, REGIONSERVER => 'all', THROTTLE_TYPE => WRITE, + LIMIT => '2req/sec', ALLOW_EXCEED => true) output = capture_stdout{ command(:list_quotas) } assert(output.include?('REGIONSERVER => all')) assert(output.include?('TYPE => THROTTLE')) - assert(output.include?('THROTTLE_TYPE => WRITE')) - assert(output.include?('THROTTLE_TYPE => REQUEST_CAPACITY_UNIT')) + assert(output.include?('THROTTLE_TYPE => WRITE_NUMBER')) assert(output.include?('LIMIT => 2req/sec')) + assert(output.include?('ALLOW_EXCEED => true')) command(:set_quota, TYPE => THROTTLE, REGIONSERVER => 'all', LIMIT => NONE) output = capture_stdout{ command(:list_quotas) } -- 2.7.4