public synchronized void maybeThrottle ( int eventsSeen ) { if ( maxRatePerSecond > 0 ) { long now = time . milliseconds ( ) ; try { rateSensor . record ( eventsSeen , now ) ; } catch ( QuotaViolationException e ) { // If we're over quota, we calculate how long to sleep to compensate. double currentRate = e . getValue ( ) ; if ( currentRate > this . maxRatePerSecond ) { double excessRate = currentRate - this . maxRatePerSecond ; long sleepTimeMs = Math . round ( excessRate / this . maxRatePerSecond * voldemort . utils . Time . MS_PER_SECOND ) ; if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Throttler quota exceeded:\n" + "eventsSeen \t= " + eventsSeen + " in this call of maybeThrotte(),\n" + "currentRate \t= " + currentRate + " events/sec,\n" + "maxRatePerSecond \t= " + this . maxRatePerSecond + " events/sec,\n" + "excessRate \t= " + excessRate + " events/sec,\n" + "sleeping for \t" + sleepTimeMs + " ms to compensate.\n" + "rateConfig.timeWindowMs() = " + rateConfig . timeWindowMs ( ) ) ; } if ( sleepTimeMs > rateConfig . timeWindowMs ( ) ) { logger . warn ( "Throttler sleep time (" + sleepTimeMs + " ms) exceeds " + "window size (" + rateConfig . timeWindowMs ( ) + " ms). This will likely " + "result in not being able to honor the rate limit accurately." ) ; // When using the HDFS Fetcher, setting the hdfs.fetcher.buffer.size // too high could cause this problem. } time . sleep ( sleepTimeMs ) ; } else if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Weird. Got QuotaValidationException but measured rate not over rateLimit: " + "currentRate = " + currentRate + " , rateLimit = " + this . maxRatePerSecond ) ; } } } }
Sleeps if necessary to slow down the caller .