[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)

[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