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: us...@infra.apache.org
