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