You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2020/07/24 09:55:32 UTC
[GitHub] [kafka] dajac opened a new pull request #9072: KAFKA-10162; Make the rate based quota behave more like a Token Bucket (KIP-599, Part III)
dajac opened a new pull request #9072:
URL: https://github.com/apache/kafka/pull/9072
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] apovzner commented on a change in pull request #9072: KAFKA-10162; Make the rate based quota behave more like a Token Bucket (KIP-599, Part III)
Posted by GitBox <gi...@apache.org>.
apovzner commented on a change in pull request #9072:
URL: https://github.com/apache/kafka/pull/9072#discussion_r461297706
##########
File path: clients/src/main/java/org/apache/kafka/common/metrics/stats/TokenBucket.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.kafka.common.metrics.stats;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.common.metrics.MetricConfig;
+
+public class TokenBucket extends SampledStat {
+
+ private final TimeUnit unit;
+
+ public TokenBucket() {
+ this(TimeUnit.SECONDS);
+ }
+
+ public TokenBucket(TimeUnit unit) {
Review comment:
Would be useful to add javadoc here for `unit` param. I understand, the unit needs to match Quota representation in `config` param passed to `record()` method, right?
##########
File path: clients/src/main/java/org/apache/kafka/common/metrics/stats/TokenBucket.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.kafka.common.metrics.stats;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.common.metrics.MetricConfig;
+
+public class TokenBucket extends SampledStat {
+
+ private final TimeUnit unit;
+
+ public TokenBucket() {
+ this(TimeUnit.SECONDS);
+ }
+
+ public TokenBucket(TimeUnit unit) {
+ super(0);
+ this.unit = unit;
+ }
+
+ @Override
+ public void record(MetricConfig config, double value, long timeMs) {
+ if (value < 0) {
+ unrecord(config, -value, timeMs);
+ return;
+ }
+
+ final double quota = quota(config);
+ final long firstTimeWindowMs = firstTimeWindowMs(config, timeMs);
+
+ // Get current sample or create one if empty
+ Sample sample = current(firstTimeWindowMs);
+
+ // Verify that the current sample was not reinitialized. If it was the case,
+ // restart from the first time window
+ if (sample.eventCount == 0) {
+ sample.reset(firstTimeWindowMs);
+ }
+
+ // Add the value to the current sample
+ sample.value += value;
+ sample.eventCount += 1;
+
+ // If current sample is completed AND a new one can be created,
+ // create one and spill over the amount above the quota to the
+ // new sample. Repeat until either the sample is not complete
+ // or no new sample can be created.
+ while (sample.isComplete(timeMs, config)) {
+ double extra = sample.value - quota;
+ sample.value = quota;
+
+ sample = advance(config, sample.lastWindowMs + config.timeWindowMs());
+ sample.value = extra;
+ sample.eventCount += 1;
+ }
+ }
+
+ private void unrecord(MetricConfig config, double value, long timeMs) {
+ final double quota = quota(config);
+ final long firstTimeWindowMs = firstTimeWindowMs(config, timeMs);
+
+ // Rewind
+ while (value > 0) {
+ // Get current sample or create one if empty
+ Sample sample = current(firstTimeWindowMs);
+
+ // If the current sample has been purged, we can't unrecord anything
+ if (sample.eventCount == 0) {
+ return;
+ }
+
+ if (value <= sample.value) {
+ sample.value -= value;
+ return;
+ }
+
+ sample.reset(timeMs);
+ value -= quota;
+
+ this.current = this.current - 1;
+ if (this.current <= 0)
+ this.current = this.samples.size() - 1;
+ }
+ }
+
+ private static double quota(MetricConfig config) {
+ if (config.quota() == null)
+ throw new IllegalStateException("TokenBucket can't be used without a quota.");
+
+ return config.quota().bound();
Review comment:
I think this should return `config.quota().bound() * convert(config.timeWindowMs())`, because we are using the value returned from this method to compare to a value in a sample window. Also could be useful to add one unit test where MetricConfig.timeWindowMs() is not 1 second.
##########
File path: clients/src/main/java/org/apache/kafka/common/metrics/stats/TokenBucket.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.kafka.common.metrics.stats;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.common.metrics.MetricConfig;
+
+public class TokenBucket extends SampledStat {
+
+ private final TimeUnit unit;
+
+ public TokenBucket() {
+ this(TimeUnit.SECONDS);
+ }
+
+ public TokenBucket(TimeUnit unit) {
+ super(0);
+ this.unit = unit;
+ }
+
+ @Override
+ public void record(MetricConfig config, double value, long timeMs) {
+ if (value < 0) {
+ unrecord(config, -value, timeMs);
+ return;
+ }
+
+ final double quota = quota(config);
+ final long firstTimeWindowMs = firstTimeWindowMs(config, timeMs);
+
+ // Get current sample or create one if empty
+ Sample sample = current(firstTimeWindowMs);
+
+ // Verify that the current sample was not reinitialized. If it was the case,
+ // restart from the first time window
+ if (sample.eventCount == 0) {
+ sample.reset(firstTimeWindowMs);
+ }
+
+ // Add the value to the current sample
+ sample.value += value;
+ sample.eventCount += 1;
+
+ // If current sample is completed AND a new one can be created,
+ // create one and spill over the amount above the quota to the
+ // new sample. Repeat until either the sample is not complete
+ // or no new sample can be created.
+ while (sample.isComplete(timeMs, config)) {
+ double extra = sample.value - quota;
+ sample.value = quota;
+
+ sample = advance(config, sample.lastWindowMs + config.timeWindowMs());
+ sample.value = extra;
+ sample.eventCount += 1;
+ }
+ }
+
+ private void unrecord(MetricConfig config, double value, long timeMs) {
+ final double quota = quota(config);
+ final long firstTimeWindowMs = firstTimeWindowMs(config, timeMs);
+
+ // Rewind
+ while (value > 0) {
+ // Get current sample or create one if empty
+ Sample sample = current(firstTimeWindowMs);
+
+ // If the current sample has been purged, we can't unrecord anything
+ if (sample.eventCount == 0) {
+ return;
+ }
+
+ if (value <= sample.value) {
+ sample.value -= value;
+ return;
+ }
+
+ sample.reset(timeMs);
+ value -= quota;
+
+ this.current = this.current - 1;
+ if (this.current <= 0)
+ this.current = this.samples.size() - 1;
+ }
+ }
+
+ private static double quota(MetricConfig config) {
+ if (config.quota() == null)
+ throw new IllegalStateException("TokenBucket can't be used without a quota.");
+
+ return config.quota().bound();
Review comment:
and maybe consider renaming it to `sampleQuota` or `sampleWindowQuota` or something that hints that this method returns quota over a sample time window?
##########
File path: clients/src/main/java/org/apache/kafka/common/metrics/stats/TokenBucket.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.kafka.common.metrics.stats;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.common.metrics.MetricConfig;
+
+public class TokenBucket extends SampledStat {
+
+ private final TimeUnit unit;
+
+ public TokenBucket() {
+ this(TimeUnit.SECONDS);
+ }
+
+ public TokenBucket(TimeUnit unit) {
Review comment:
This should be the same unit as the unit in `Rate`, right? If so, I think someone could create Rate as:
```new Rate(TimeUnit.MILLISECONDS, new TokenBucket())```
Or
```new Rate(new TokenBucket(TimeUnit.MILLISECONDS))```
##########
File path: clients/src/main/java/org/apache/kafka/common/metrics/stats/TokenBucket.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.kafka.common.metrics.stats;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.common.metrics.MetricConfig;
+
+public class TokenBucket extends SampledStat {
+
+ private final TimeUnit unit;
+
+ public TokenBucket() {
+ this(TimeUnit.SECONDS);
+ }
+
+ public TokenBucket(TimeUnit unit) {
+ super(0);
+ this.unit = unit;
+ }
+
+ @Override
+ public void record(MetricConfig config, double value, long timeMs) {
+ if (value < 0) {
+ unrecord(config, -value, timeMs);
+ return;
+ }
+
+ final double quota = quota(config);
+ final long firstTimeWindowMs = firstTimeWindowMs(config, timeMs);
+
+ // Get current sample or create one if empty
+ Sample sample = current(firstTimeWindowMs);
+
+ // Verify that the current sample was not reinitialized. If it was the case,
+ // restart from the first time window
+ if (sample.eventCount == 0) {
+ sample.reset(firstTimeWindowMs);
+ }
+
+ // Add the value to the current sample
+ sample.value += value;
+ sample.eventCount += 1;
+
+ // If current sample is completed AND a new one can be created,
+ // create one and spill over the amount above the quota to the
+ // new sample. Repeat until either the sample is not complete
+ // or no new sample can be created.
+ while (sample.isComplete(timeMs, config)) {
+ double extra = sample.value - quota;
+ sample.value = quota;
+
+ sample = advance(config, sample.lastWindowMs + config.timeWindowMs());
+ sample.value = extra;
+ sample.eventCount += 1;
+ }
+ }
+
+ private void unrecord(MetricConfig config, double value, long timeMs) {
+ final double quota = quota(config);
+ final long firstTimeWindowMs = firstTimeWindowMs(config, timeMs);
+
+ // Rewind
+ while (value > 0) {
+ // Get current sample or create one if empty
+ Sample sample = current(firstTimeWindowMs);
+
+ // If the current sample has been purged, we can't unrecord anything
+ if (sample.eventCount == 0) {
+ return;
+ }
+
+ if (value <= sample.value) {
+ sample.value -= value;
+ return;
+ }
+
+ sample.reset(timeMs);
+ value -= quota;
+
+ this.current = this.current - 1;
+ if (this.current <= 0)
Review comment:
should this be `< 0`? Otherwise, we skip index 0
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac closed pull request #9072: KAFKA-10162; Make the rate based quota behave more like a Token Bucket (KIP-599, Part III)
Posted by GitBox <gi...@apache.org>.
dajac closed pull request #9072:
URL: https://github.com/apache/kafka/pull/9072
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a change in pull request #9072: KAFKA-10162; Make the rate based quota behave more like a Token Bucket (KIP-599, Part III)
Posted by GitBox <gi...@apache.org>.
junrao commented on a change in pull request #9072:
URL: https://github.com/apache/kafka/pull/9072#discussion_r463336327
##########
File path: clients/src/main/java/org/apache/kafka/common/metrics/stats/TokenBucket.java
##########
@@ -0,0 +1,179 @@
+/*
+ * 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.kafka.common.metrics.stats;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.common.metrics.MetricConfig;
+
+/**
+ * The {@link TokenBucket} is a {@link SampledStat} implementing a token bucket that can be used
+ * in conjunction with a {@link Rate} to enforce a quota.
+ *
+ * A token bucket accumulates tokens with a constant rate R, one token per 1/R second, up to a
+ * maximum burst size B. The burst size essentially means that we keep permission to do a unit of
+ * work for up to B / R seconds.
+ *
+ * The {@link TokenBucket} adapts this to fit within a {@link SampledStat}. It accumulates tokens
+ * in chunks of Q units by default (sample length * quota) instead of one token every 1/R second
+ * and expires in chunks of Q units too (when the oldest sample is expired).
+ *
+ * Internally, we achieve this behavior by not completing the current sample until we fill that
+ * sample up to Q (used all credits). Samples are filled up one after the others until the maximum
+ * number of samples is reached. If it is not possible to created a new sample, we accumulate in
+ * the last one until a new one can be created. The over used credits are spilled over to the new
+ * sample at when it is created. Every time a sample is purged, Q credits are made available.
+ *
+ * It is important to note that the maximum burst is not enforced in the class and depends on
+ * how the quota is enforced in the {@link Rate}.
+ */
+public class TokenBucket extends SampledStat {
+
+ private final TimeUnit unit;
+
+ /**
+ * Instantiates a new TokenBucket that works by default with a Quota {@link MetricConfig#quota()}
+ * in {@link TimeUnit#SECONDS}.
+ */
+ public TokenBucket() {
+ this(TimeUnit.SECONDS);
+ }
+
+ /**
+ * Instantiates a new TokenBucket that works with the provided time unit.
+ *
+ * @param unit The time unit of the Quota {@link MetricConfig#quota()}
+ */
+ public TokenBucket(TimeUnit unit) {
+ super(0);
+ this.unit = unit;
+ }
+
+ @Override
+ public void record(MetricConfig config, double value, long timeMs) {
Review comment:
@apovzner @dajac : Currently, we calculate the delayed time based on QuotaViolationException.value and QuotaViolationException.bound. I was thinking that we could pass some additional info in QuotaViolationException so that we could calculate the delayed time properly.
Overall, it seems that decoupling the observation from token bucket based quota might be easier to understand. As for monitoring the remaining credit, we could add a separate metric.
Also, it seems that quota only makes sense for rates. So, instead of making quota available on arbitrary measurable, we could just make it work for Rate.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] apovzner commented on a change in pull request #9072: KAFKA-10162; Make the rate based quota behave more like a Token Bucket (KIP-599, Part III)
Posted by GitBox <gi...@apache.org>.
apovzner commented on a change in pull request #9072:
URL: https://github.com/apache/kafka/pull/9072#discussion_r462645570
##########
File path: clients/src/main/java/org/apache/kafka/common/metrics/stats/TokenBucket.java
##########
@@ -0,0 +1,179 @@
+/*
+ * 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.kafka.common.metrics.stats;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.common.metrics.MetricConfig;
+
+/**
+ * The {@link TokenBucket} is a {@link SampledStat} implementing a token bucket that can be used
+ * in conjunction with a {@link Rate} to enforce a quota.
+ *
+ * A token bucket accumulates tokens with a constant rate R, one token per 1/R second, up to a
+ * maximum burst size B. The burst size essentially means that we keep permission to do a unit of
+ * work for up to B / R seconds.
+ *
+ * The {@link TokenBucket} adapts this to fit within a {@link SampledStat}. It accumulates tokens
+ * in chunks of Q units by default (sample length * quota) instead of one token every 1/R second
+ * and expires in chunks of Q units too (when the oldest sample is expired).
+ *
+ * Internally, we achieve this behavior by not completing the current sample until we fill that
+ * sample up to Q (used all credits). Samples are filled up one after the others until the maximum
+ * number of samples is reached. If it is not possible to created a new sample, we accumulate in
+ * the last one until a new one can be created. The over used credits are spilled over to the new
+ * sample at when it is created. Every time a sample is purged, Q credits are made available.
+ *
+ * It is important to note that the maximum burst is not enforced in the class and depends on
+ * how the quota is enforced in the {@link Rate}.
+ */
+public class TokenBucket extends SampledStat {
+
+ private final TimeUnit unit;
+
+ /**
+ * Instantiates a new TokenBucket that works by default with a Quota {@link MetricConfig#quota()}
+ * in {@link TimeUnit#SECONDS}.
+ */
+ public TokenBucket() {
+ this(TimeUnit.SECONDS);
+ }
+
+ /**
+ * Instantiates a new TokenBucket that works with the provided time unit.
+ *
+ * @param unit The time unit of the Quota {@link MetricConfig#quota()}
+ */
+ public TokenBucket(TimeUnit unit) {
+ super(0);
+ this.unit = unit;
+ }
+
+ @Override
+ public void record(MetricConfig config, double value, long timeMs) {
Review comment:
@junrao Regarding "With this change, if we record a large value, the observed effect of the value could last much longer than the number of samples. " -- This will not happen with this approach. If we record a very large value, we never move to the bucket with timestamp > current timestamp (of the recording time). This approach can only add the value to older buckets, which did not expire, but never to the buckets "in the future".
For example, if we only have 2 samples, and perSampleQuota = 5, and say we already filled in both buckets up to quota: [5, 5]. If new requests arrive but the timestamp did not move past the last bucket, we are going to be adding this value to the last bucket, for example getting to [5, 20] if we recorded 15. If the next recording happens after the time moved passed the last bucket, say we record 3, then buckets will look like [20, 3].
##########
File path: clients/src/main/java/org/apache/kafka/common/metrics/stats/TokenBucket.java
##########
@@ -0,0 +1,179 @@
+/*
+ * 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.kafka.common.metrics.stats;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.common.metrics.MetricConfig;
+
+/**
+ * The {@link TokenBucket} is a {@link SampledStat} implementing a token bucket that can be used
+ * in conjunction with a {@link Rate} to enforce a quota.
+ *
+ * A token bucket accumulates tokens with a constant rate R, one token per 1/R second, up to a
+ * maximum burst size B. The burst size essentially means that we keep permission to do a unit of
+ * work for up to B / R seconds.
+ *
+ * The {@link TokenBucket} adapts this to fit within a {@link SampledStat}. It accumulates tokens
+ * in chunks of Q units by default (sample length * quota) instead of one token every 1/R second
+ * and expires in chunks of Q units too (when the oldest sample is expired).
+ *
+ * Internally, we achieve this behavior by not completing the current sample until we fill that
+ * sample up to Q (used all credits). Samples are filled up one after the others until the maximum
+ * number of samples is reached. If it is not possible to created a new sample, we accumulate in
+ * the last one until a new one can be created. The over used credits are spilled over to the new
+ * sample at when it is created. Every time a sample is purged, Q credits are made available.
+ *
+ * It is important to note that the maximum burst is not enforced in the class and depends on
+ * how the quota is enforced in the {@link Rate}.
+ */
+public class TokenBucket extends SampledStat {
+
+ private final TimeUnit unit;
+
+ /**
+ * Instantiates a new TokenBucket that works by default with a Quota {@link MetricConfig#quota()}
+ * in {@link TimeUnit#SECONDS}.
+ */
+ public TokenBucket() {
+ this(TimeUnit.SECONDS);
+ }
+
+ /**
+ * Instantiates a new TokenBucket that works with the provided time unit.
+ *
+ * @param unit The time unit of the Quota {@link MetricConfig#quota()}
+ */
+ public TokenBucket(TimeUnit unit) {
+ super(0);
+ this.unit = unit;
+ }
+
+ @Override
+ public void record(MetricConfig config, double value, long timeMs) {
Review comment:
@junrao If I understood your proposal correctly, we will keep Rate calculation the same, but additionally implement TokenBucket (traditional way) which will tell us when quota is violated. This would be much easier. However, I think, it would not fix our issue (that we are trying to fix) of too large throttle times during bursty workload. ClientQuotaManager calculates throttle times by comparing rate (based on how we record Rate) to quota, which I think would result in the same behavior as before unless we change the way we calculate throttle time as well.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a change in pull request #9072: KAFKA-10162; Make the rate based quota behave more like a Token Bucket (KIP-599, Part III)
Posted by GitBox <gi...@apache.org>.
junrao commented on a change in pull request #9072:
URL: https://github.com/apache/kafka/pull/9072#discussion_r462624834
##########
File path: clients/src/main/java/org/apache/kafka/common/metrics/stats/TokenBucket.java
##########
@@ -0,0 +1,179 @@
+/*
+ * 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.kafka.common.metrics.stats;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.common.metrics.MetricConfig;
+
+/**
+ * The {@link TokenBucket} is a {@link SampledStat} implementing a token bucket that can be used
+ * in conjunction with a {@link Rate} to enforce a quota.
+ *
+ * A token bucket accumulates tokens with a constant rate R, one token per 1/R second, up to a
+ * maximum burst size B. The burst size essentially means that we keep permission to do a unit of
+ * work for up to B / R seconds.
+ *
+ * The {@link TokenBucket} adapts this to fit within a {@link SampledStat}. It accumulates tokens
+ * in chunks of Q units by default (sample length * quota) instead of one token every 1/R second
+ * and expires in chunks of Q units too (when the oldest sample is expired).
+ *
+ * Internally, we achieve this behavior by not completing the current sample until we fill that
+ * sample up to Q (used all credits). Samples are filled up one after the others until the maximum
+ * number of samples is reached. If it is not possible to created a new sample, we accumulate in
+ * the last one until a new one can be created. The over used credits are spilled over to the new
+ * sample at when it is created. Every time a sample is purged, Q credits are made available.
+ *
+ * It is important to note that the maximum burst is not enforced in the class and depends on
+ * how the quota is enforced in the {@link Rate}.
+ */
+public class TokenBucket extends SampledStat {
+
+ private final TimeUnit unit;
+
+ /**
+ * Instantiates a new TokenBucket that works by default with a Quota {@link MetricConfig#quota()}
+ * in {@link TimeUnit#SECONDS}.
+ */
+ public TokenBucket() {
+ this(TimeUnit.SECONDS);
+ }
+
+ /**
+ * Instantiates a new TokenBucket that works with the provided time unit.
+ *
+ * @param unit The time unit of the Quota {@link MetricConfig#quota()}
+ */
+ public TokenBucket(TimeUnit unit) {
+ super(0);
+ this.unit = unit;
+ }
+
+ @Override
+ public void record(MetricConfig config, double value, long timeMs) {
Review comment:
Just a high level comment on the approach. This approach tries to spread a recorded value to multiple samples if the sample level quota is exceeded. While this matches the token bucket behavior for quota, it changes the behavior when we observe the value of the measurable. Currently, we record a full value in the current Sample. When we observe the value of the measurable, the effect of this value will last for the number of samples. After which, this value rolls out and no longer impacts the observed measurable. With this change, if we record a large value, the observed effect of the value could last much longer than the number of samples. For example, if we have 10 1-sec samples each with a quota of 1 and we record a value of 1000, the effect of this value will last for 1000 secs instead of 10 secs from the observability perspective. This may cause some confusion since we don't quite know when an event actually occurred.
I was thinking about an alternative approach that decouples the recording/observation of the measurable from quota calculation. Here is the rough idea. We create a customized SampledStat that records new values in a single sample as it is. In addition, it maintains an accumulated available credit. As time advances, we add new credits based on the quota rate, capped by samples * perSampleQuota. When a value is recorded, we deduct the value from the credit and allow the credit to go below 0. We change the Sensor.checkQuotas() logic such that if the customized SampledStat is used, we throw QuotaViolationException if credit is < 0. This preserves the current behavior for observability, but allows quota to be enforced based on token bucket.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #9072: KAFKA-10162; Make the rate based quota behave more like a Token Bucket (KIP-599, Part III)
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #9072:
URL: https://github.com/apache/kafka/pull/9072#issuecomment-664923691
@apovzner Thanks for your comments. I have updated the PR to incorporate your feedback. I have also fixed a bug in unrecord and added the javadoc.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #9072: KAFKA-10162; Make the rate based quota behave more like a Token Bucket (KIP-599, Part III)
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #9072:
URL: https://github.com/apache/kafka/pull/9072#discussion_r461123154
##########
File path: clients/src/main/java/org/apache/kafka/common/metrics/stats/TokenBucket.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.kafka.common.metrics.stats;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.common.metrics.MetricConfig;
+
+public class TokenBucket extends SampledStat {
Review comment:
Note to myself: Add javadoc.
##########
File path: clients/src/main/java/org/apache/kafka/common/metrics/stats/TokenBucket.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.kafka.common.metrics.stats;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.common.metrics.MetricConfig;
+
+public class TokenBucket extends SampledStat {
+
+ private final TimeUnit unit;
+
+ public TokenBucket() {
+ this(TimeUnit.SECONDS);
+ }
+
+ public TokenBucket(TimeUnit unit) {
Review comment:
That is correct. This is bit unfortunate but as the two are independent from each others, we can't enforce using the same timeunit nor reuse the time unit of the rate.
I will make this clear in the javadoc.
##########
File path: clients/src/main/java/org/apache/kafka/common/metrics/stats/TokenBucket.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.kafka.common.metrics.stats;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.common.metrics.MetricConfig;
+
+public class TokenBucket extends SampledStat {
+
+ private final TimeUnit unit;
+
+ public TokenBucket() {
+ this(TimeUnit.SECONDS);
+ }
+
+ public TokenBucket(TimeUnit unit) {
+ super(0);
+ this.unit = unit;
+ }
+
+ @Override
+ public void record(MetricConfig config, double value, long timeMs) {
+ if (value < 0) {
+ unrecord(config, -value, timeMs);
+ return;
+ }
+
+ final double quota = quota(config);
+ final long firstTimeWindowMs = firstTimeWindowMs(config, timeMs);
+
+ // Get current sample or create one if empty
+ Sample sample = current(firstTimeWindowMs);
+
+ // Verify that the current sample was not reinitialized. If it was the case,
+ // restart from the first time window
+ if (sample.eventCount == 0) {
+ sample.reset(firstTimeWindowMs);
+ }
+
+ // Add the value to the current sample
+ sample.value += value;
+ sample.eventCount += 1;
+
+ // If current sample is completed AND a new one can be created,
+ // create one and spill over the amount above the quota to the
+ // new sample. Repeat until either the sample is not complete
+ // or no new sample can be created.
+ while (sample.isComplete(timeMs, config)) {
+ double extra = sample.value - quota;
+ sample.value = quota;
+
+ sample = advance(config, sample.lastWindowMs + config.timeWindowMs());
+ sample.value = extra;
+ sample.eventCount += 1;
+ }
+ }
+
+ private void unrecord(MetricConfig config, double value, long timeMs) {
+ final double quota = quota(config);
+ final long firstTimeWindowMs = firstTimeWindowMs(config, timeMs);
+
+ // Rewind
+ while (value > 0) {
+ // Get current sample or create one if empty
+ Sample sample = current(firstTimeWindowMs);
+
+ // If the current sample has been purged, we can't unrecord anything
+ if (sample.eventCount == 0) {
+ return;
+ }
+
+ if (value <= sample.value) {
+ sample.value -= value;
+ return;
+ }
+
+ sample.reset(timeMs);
+ value -= quota;
+
+ this.current = this.current - 1;
+ if (this.current <= 0)
Review comment:
Good catch!
##########
File path: clients/src/main/java/org/apache/kafka/common/metrics/stats/TokenBucket.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.kafka.common.metrics.stats;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.common.metrics.MetricConfig;
+
+public class TokenBucket extends SampledStat {
+
+ private final TimeUnit unit;
+
+ public TokenBucket() {
+ this(TimeUnit.SECONDS);
+ }
+
+ public TokenBucket(TimeUnit unit) {
+ super(0);
+ this.unit = unit;
+ }
+
+ @Override
+ public void record(MetricConfig config, double value, long timeMs) {
+ if (value < 0) {
+ unrecord(config, -value, timeMs);
+ return;
+ }
+
+ final double quota = quota(config);
+ final long firstTimeWindowMs = firstTimeWindowMs(config, timeMs);
+
+ // Get current sample or create one if empty
+ Sample sample = current(firstTimeWindowMs);
+
+ // Verify that the current sample was not reinitialized. If it was the case,
+ // restart from the first time window
+ if (sample.eventCount == 0) {
+ sample.reset(firstTimeWindowMs);
+ }
+
+ // Add the value to the current sample
+ sample.value += value;
+ sample.eventCount += 1;
+
+ // If current sample is completed AND a new one can be created,
+ // create one and spill over the amount above the quota to the
+ // new sample. Repeat until either the sample is not complete
+ // or no new sample can be created.
+ while (sample.isComplete(timeMs, config)) {
+ double extra = sample.value - quota;
+ sample.value = quota;
+
+ sample = advance(config, sample.lastWindowMs + config.timeWindowMs());
+ sample.value = extra;
+ sample.eventCount += 1;
+ }
+ }
+
+ private void unrecord(MetricConfig config, double value, long timeMs) {
+ final double quota = quota(config);
+ final long firstTimeWindowMs = firstTimeWindowMs(config, timeMs);
+
+ // Rewind
+ while (value > 0) {
+ // Get current sample or create one if empty
+ Sample sample = current(firstTimeWindowMs);
+
+ // If the current sample has been purged, we can't unrecord anything
+ if (sample.eventCount == 0) {
+ return;
+ }
+
+ if (value <= sample.value) {
+ sample.value -= value;
+ return;
+ }
+
+ sample.reset(timeMs);
+ value -= quota;
+
+ this.current = this.current - 1;
+ if (this.current <= 0)
+ this.current = this.samples.size() - 1;
+ }
+ }
+
+ private static double quota(MetricConfig config) {
+ if (config.quota() == null)
+ throw new IllegalStateException("TokenBucket can't be used without a quota.");
+
+ return config.quota().bound();
Review comment:
You're absolutely right.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #9072: KAFKA-10162; Make the rate based quota behave more like a Token Bucket (KIP-599, Part III)
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #9072:
URL: https://github.com/apache/kafka/pull/9072#discussion_r462846252
##########
File path: clients/src/main/java/org/apache/kafka/common/metrics/stats/TokenBucket.java
##########
@@ -0,0 +1,179 @@
+/*
+ * 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.kafka.common.metrics.stats;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.common.metrics.MetricConfig;
+
+/**
+ * The {@link TokenBucket} is a {@link SampledStat} implementing a token bucket that can be used
+ * in conjunction with a {@link Rate} to enforce a quota.
+ *
+ * A token bucket accumulates tokens with a constant rate R, one token per 1/R second, up to a
+ * maximum burst size B. The burst size essentially means that we keep permission to do a unit of
+ * work for up to B / R seconds.
+ *
+ * The {@link TokenBucket} adapts this to fit within a {@link SampledStat}. It accumulates tokens
+ * in chunks of Q units by default (sample length * quota) instead of one token every 1/R second
+ * and expires in chunks of Q units too (when the oldest sample is expired).
+ *
+ * Internally, we achieve this behavior by not completing the current sample until we fill that
+ * sample up to Q (used all credits). Samples are filled up one after the others until the maximum
+ * number of samples is reached. If it is not possible to created a new sample, we accumulate in
+ * the last one until a new one can be created. The over used credits are spilled over to the new
+ * sample at when it is created. Every time a sample is purged, Q credits are made available.
+ *
+ * It is important to note that the maximum burst is not enforced in the class and depends on
+ * how the quota is enforced in the {@link Rate}.
+ */
+public class TokenBucket extends SampledStat {
+
+ private final TimeUnit unit;
+
+ /**
+ * Instantiates a new TokenBucket that works by default with a Quota {@link MetricConfig#quota()}
+ * in {@link TimeUnit#SECONDS}.
+ */
+ public TokenBucket() {
+ this(TimeUnit.SECONDS);
+ }
+
+ /**
+ * Instantiates a new TokenBucket that works with the provided time unit.
+ *
+ * @param unit The time unit of the Quota {@link MetricConfig#quota()}
+ */
+ public TokenBucket(TimeUnit unit) {
+ super(0);
+ this.unit = unit;
+ }
+
+ @Override
+ public void record(MetricConfig config, double value, long timeMs) {
Review comment:
@junrao Thanks for your comment. Your observation is correct. This is because we spill over the amount above the quota into any newly created sample. So @apovzner's example would actually end up like this: [5, 18]. @apovzner I think that you have mentioned once another way to do this that may not require to spill over the remainder.
Your remark regarding the observability is right. This is true as soon as we deviate from a sampled rate, regardless of the implementation that we may choose eventually. Separating concerns is indeed a good idea if we can. Regarding your suggested approach, I think that may works. As @apovzner said, we will need to change the way the throttle time is computed so that it does not rely on the rate but on the amount of credits. Another concern is that the credits won't be observable neither so we may keep the correct rate but still may not understand why one is throttled or not if we can't observe the amount of credits used/left.
If we want to separate concerns, having both a sampled Rate and a Token Bucket working side by side would be the best. The Rate would continue to provide the actual Rate as of today and the Token Bucket would be used to enforced the quota. We could expose the amount of credits in the bucket via a new metric. One way to achieve this would be to have two MeasurableStats within the Sensor: the Rate and the Token Bucket. We would still need to have an adapted version of the Token Bucket that works with our current configs (quota, samples, window). I implemented one to compare with the current approach few weeks ago: https://github.com/dajac/kafka/blob/19e7180105c366d245183644c823135fb2872571/clients/src/main/java/org/apache/kafka/common/metrics/stats/RealTokenBucket.java. This one works within a Rate though.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org