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