001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.quotas; 019 020import org.apache.hadoop.conf.Configuration; 021import org.apache.yetus.audience.InterfaceAudience; 022import org.apache.yetus.audience.InterfaceStability; 023import org.slf4j.Logger; 024import org.slf4j.LoggerFactory; 025 026import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos; 027 028/* 029 * Internal class used to check and consume quota if exceed throttle quota is enabled. Exceed 030 * throttle quota means, user can over consume user/namespace/table quota if region server has 031 * additional available quota because other users don't consume at the same time. 032 * 033 * There are some limits when enable exceed throttle quota: 034 * 1. Must set at least one read and one write region server throttle quota; 035 * 2. All region server throttle quotas must be in seconds time unit. Because once previous requests 036 * exceed their quota and consume region server quota, quota in other time units may be refilled in 037 * a long time, this may affect later requests. 038 */ 039@InterfaceAudience.Private 040@InterfaceStability.Evolving 041public class ExceedOperationQuota extends DefaultOperationQuota { 042 private static final Logger LOG = LoggerFactory.getLogger(ExceedOperationQuota.class); 043 private QuotaLimiter regionServerLimiter; 044 045 public ExceedOperationQuota(final Configuration conf, int blockSizeBytes, 046 QuotaLimiter regionServerLimiter, final QuotaLimiter... limiters) { 047 super(conf, blockSizeBytes, limiters); 048 this.regionServerLimiter = regionServerLimiter; 049 } 050 051 @Override 052 public void checkBatchQuota(int numWrites, int numReads, boolean isAtomic) 053 throws RpcThrottlingException { 054 Runnable estimateQuota = () -> updateEstimateConsumeBatchQuota(numWrites, numReads); 055 CheckQuotaRunnable checkQuota = () -> super.checkBatchQuota(numWrites, numReads, isAtomic); 056 checkQuota(estimateQuota, checkQuota, numWrites, numReads, 0, isAtomic); 057 } 058 059 @Override 060 public void checkScanQuota(ClientProtos.ScanRequest scanRequest, long maxScannerResultSize, 061 long maxBlockBytesScanned, long prevBlockBytesScannedDifference) throws RpcThrottlingException { 062 Runnable estimateQuota = () -> updateEstimateConsumeScanQuota(scanRequest, maxScannerResultSize, 063 maxBlockBytesScanned, prevBlockBytesScannedDifference); 064 CheckQuotaRunnable checkQuota = () -> super.checkScanQuota(scanRequest, maxScannerResultSize, 065 maxBlockBytesScanned, prevBlockBytesScannedDifference); 066 checkQuota(estimateQuota, checkQuota, 0, 0, 1, false); 067 } 068 069 private void checkQuota(Runnable estimateQuota, CheckQuotaRunnable checkQuota, int numWrites, 070 int numReads, int numScans, boolean isAtomic) throws RpcThrottlingException { 071 if (regionServerLimiter.isBypass()) { 072 // If region server limiter is bypass, which means no region server quota is set, check and 073 // throttle by all other quotas. In this condition, exceed throttle quota will not work. 074 LOG.warn("Exceed throttle quota is enabled but no region server quotas found"); 075 checkQuota.run(); 076 } else { 077 // 1. Update estimate quota which will be consumed 078 estimateQuota.run(); 079 // 2. Check if region server limiter is enough. If not, throw RpcThrottlingException. 080 regionServerLimiter.checkQuota(numWrites, writeConsumed, numReads + numScans, readConsumed, 081 writeCapacityUnitConsumed, readCapacityUnitConsumed, isAtomic); 082 // 3. Check if other limiters are enough. If not, exceed other limiters because region server 083 // limiter is enough. 084 boolean exceed = false; 085 try { 086 checkQuota.run(); 087 } catch (RpcThrottlingException e) { 088 exceed = true; 089 if (LOG.isDebugEnabled()) { 090 LOG.debug("Read/Write requests num exceeds quota: writes:{} reads:{}, scans:{}, " 091 + "try use region server quota", numWrites, numReads, numScans); 092 } 093 } 094 // 4. Region server limiter is enough and grab estimated consume quota. 095 readAvailable = Math.max(readAvailable, regionServerLimiter.getReadAvailable()); 096 regionServerLimiter.grabQuota(numWrites, writeConsumed, numReads + numScans, readConsumed, 097 writeCapacityUnitConsumed, writeCapacityUnitConsumed, isAtomic); 098 if (exceed) { 099 // 5. Other quota limiter is exceeded and has not been grabbed (because throw 100 // RpcThrottlingException in Step 3), so grab it. 101 for (final QuotaLimiter limiter : limiters) { 102 limiter.grabQuota(numWrites, writeConsumed, numReads + numScans, readConsumed, 103 writeCapacityUnitConsumed, writeCapacityUnitConsumed, isAtomic); 104 } 105 } 106 } 107 } 108 109 @Override 110 public void close() { 111 super.close(); 112 if (writeDiff != 0) { 113 regionServerLimiter.consumeWrite(writeDiff, writeCapacityUnitDiff, false); 114 } 115 if (readDiff != 0) { 116 regionServerLimiter.consumeRead(readDiff, readCapacityUnitDiff, false); 117 } 118 } 119 120 private interface CheckQuotaRunnable { 121 void run() throws RpcThrottlingException; 122 } 123}