> On Aug. 6, 2015, 2:02 a.m., Jun Rao wrote: > > clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java, lines > > 135-139 > > <https://reviews.apache.org/r/33049/diff/18/?file=1032143#file1032143line135> > > > > Is that calculation here right? Based on the calculation in Throttler, > > it should be sth like the following. > > > > metricValue/quota.bound() - windowSize > > > > Also, the window size is calculated as config.sampels() * > > config.timeWindows(), which is inaccurate. The last window is not complete. > > So we need to take the current time into consideration. > > > > Finally, it seems that delayTime only makes sense for rates and not for > > anything else. Perhaps we can at least add a comment. > > Aditya Auradkar wrote: > Hey Jun - > > Can you elaborate a little? How would we use the current time exactly? It > is not clear to me how subtracting the windowSize (time unit) from a fraction > (metricValue/quota.bound()) gives the right delay. > > I also added a comment for delayTime making sense for rates only.
I dug into the Throttler code a bit. Basically - The metricValue is the absolute actual value and not a rate. Throttler delay: (currentValueOfCounter/quotaRate) - elapsedTime Current Sensor delay: ((currentRateValue - quotaRate)/quotaRate) * elapsedTime Lets take an example (all in seconds): quotaRate = 10QPS elapedSec = 20 (Lets say 21 windows of 1 second each. The last second has not started yet) currentValueOfCounter = 250 currentRate = (250/20) = 12.5 (assuming 20 elapsed seconds. Current second may be incomplete) Throttler Formula delay = (currentValueOfCounter/quotaRate) - elapsedTime = (250/10) - 20 = 5 second delay Current Sensor delay = ((currentRateValue - quotaRate)/quotaRate) * numWindows * windowSize = ((12.5 - 10)/10) * 21 windows * 1 second window = 2.5/10 * 21 = 21/4 = 5.25 second delay I think the only discrepancy is in the "elapsedTime". The last window is not complete but should be very similar because we configure many small samples. The rate calculation is done inside Rate.java which does not expose the exact elapsed time and the actual counter value. Let's examine how the rate changes because of this: The currentRate returned by the sample will still be 12.5. However, Sensor.java uses 21 as the value because we have 21 windows configured. If we got the exact elapsed time, we would use 20 elapsed seconds Potential Sensor delay = ((currentRateValue - quotaRate)/quotaRate) * elapsedTime = ((12.5 - 10)/10) * 20 = 2.5/10 * 20 = 20/4 = 5 second delay The values are now exactly identical. Given that 5.25 second delay is slightly more pessimistic, we basically throttle a bit more aggressively if the last window is not complete. We can improve this in 2 ways: - Expose elapsedTime and raw metric value from Stat.java - A better solution (IMO), is to throw QuotaViolationException from Rate.java itself. We can also add the delayTime within the Rate class since it does not make sense for other metrics. Finally, We should consider refactoring Throttler.scala to use the Rate metric for its quota computation since the formulae will be identical and only performed in 1 place. I can tackle all of this pretty quickly in a followup patch if that is acceptable. - Aditya ----------------------------------------------------------- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/33049/#review94333 ----------------------------------------------------------- On Aug. 7, 2015, 6:28 p.m., Aditya Auradkar wrote: > > ----------------------------------------------------------- > This is an automatically generated e-mail. To reply, visit: > https://reviews.apache.org/r/33049/ > ----------------------------------------------------------- > > (Updated Aug. 7, 2015, 6:28 p.m.) > > > Review request for kafka, Joel Koshy and Jun Rao. > > > Bugs: KAFKA-2084 > https://issues.apache.org/jira/browse/KAFKA-2084 > > > Repository: kafka > > > Description > ------- > > Updated patch for quotas. This patch does the following: > 1. Add per-client metrics for both producer and consumers > 2. Add configuration for quotas > 3. Compute delay times in the metrics package and return the delay times in > QuotaViolationException > 4. Add a DelayQueue in KafkaApi's that can be used to throttle any type of > request. Implemented request throttling for produce and fetch requests. > 5. Added unit and integration test cases for both producer and consumer > 6. This doesn't include a system test. There is a separate ticket for that > 7. Fixed KAFKA-2191 - (Included fix from : > https://reviews.apache.org/r/34418/ ) > > Addressed comments from Joel and Jun > > > Diffs > ----- > > clients/src/main/java/org/apache/kafka/common/metrics/Quota.java > d82bb0c055e631425bc1ebbc7d387baac76aeeaa > > clients/src/main/java/org/apache/kafka/common/metrics/QuotaViolationException.java > a451e5385c9eca76b38b425e8ac856b2715fcffe > clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java > ca823fd4639523018311b814fde69b6177e73b97 > clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java > 98429da34418f7f1deba1b5e44e2e6025212edb3 > clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java > 544e120594de78c43581a980b1e4087b4fb98ccb > clients/src/test/java/org/apache/kafka/common/utils/MockTime.java > core/src/main/scala/kafka/server/ClientQuotaManager.scala PRE-CREATION > core/src/main/scala/kafka/server/KafkaApis.scala > 18f5b5b895af1469876b2223841fd90a2dd255e0 > core/src/main/scala/kafka/server/KafkaConfig.scala > dbe170f87331f43e2dc30165080d2cb7dfe5fdbf > core/src/main/scala/kafka/server/KafkaServer.scala > 84d4730ac634f9a5bf12a656e422fea03ad72da8 > core/src/main/scala/kafka/server/ReplicaManager.scala > 795220e7f63d163be90738b4c1a39687b44c1395 > core/src/main/scala/kafka/server/ThrottledResponse.scala PRE-CREATION > core/src/main/scala/kafka/utils/ShutdownableThread.scala > fc226c863095b7761290292cd8755cd7ad0f155c > core/src/test/scala/integration/kafka/api/QuotasTest.scala PRE-CREATION > core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala > PRE-CREATION > core/src/test/scala/unit/kafka/server/KafkaConfigConfigDefTest.scala > 04a02e08a54139ee1a298c5354731bae009efef3 > core/src/test/scala/unit/kafka/server/ThrottledResponseExpirationTest.scala > PRE-CREATION > > Diff: https://reviews.apache.org/r/33049/diff/ > > > Testing > ------- > > > Thanks, > > Aditya Auradkar > >
