> 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. > > Aditya Auradkar wrote: > 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. > > Jun Rao wrote: > Got it. I think your calculation is correct. Perhaps we can add a comment > on how the formula is derived. However, it does seem that rate includes the > last partial window in calculating the rate (see Rate.measure()). So, in your > formula, elapsedTime should probably be convert(now - > stat.oldest(now).lastWindowMs) where now is passed in from record().
Thanks Jun. I'll add a comment. You are right about the elapsedTime using the last partial window. I used 20 seconds in my example for simplicity. - Aditya ----------------------------------------------------------- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/33049/#review94333 ----------------------------------------------------------- On Aug. 10, 2015, 8:49 p.m., Aditya Auradkar wrote: > > ----------------------------------------------------------- > This is an automatically generated e-mail. To reply, visit: > https://reviews.apache.org/r/33049/ > ----------------------------------------------------------- > > (Updated Aug. 10, 2015, 8:49 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 > 7ea509c2c41acc00430c74e025e069a833aac4e7 > 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/KafkaConfigTest.scala > f32d206d3f52f3f9f4d649c213edd7058f4b6150 > core/src/test/scala/unit/kafka/server/ThrottledResponseExpirationTest.scala > PRE-CREATION > > Diff: https://reviews.apache.org/r/33049/diff/ > > > Testing > ------- > > > Thanks, > > Aditya Auradkar > >
