You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/08/22 21:10:04 UTC
(kafka) branch trunk updated: KAFKA-17372 Move `ThrottledChannelExpirationTest#testThrottledChannelDelay` and `ThrottledChannel` to server module (#16935)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 79f979cd525 KAFKA-17372 Move `ThrottledChannelExpirationTest#testThrottledChannelDelay` and `ThrottledChannel` to server module (#16935)
79f979cd525 is described below
commit 79f979cd525a168ea51148974e0d105acb0cd2b9
Author: TengYao Chi <ki...@gmail.com>
AuthorDate: Fri Aug 23 05:09:57 2024 +0800
KAFKA-17372 Move `ThrottledChannelExpirationTest#testThrottledChannelDelay` and `ThrottledChannel` to server module (#16935)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../scala/kafka/server/ClientQuotaManager.scala | 2 +-
.../scala/kafka/server/RequestHandlerHelper.scala | 1 +
.../main/scala/kafka/server/ThrottledChannel.scala | 61 ------------------
.../unit/kafka/network/SocketServerTest.scala | 1 +
.../kafka/server/BaseClientQuotaManagerTest.scala | 1 +
.../scala/unit/kafka/server/KafkaApisTest.scala | 1 +
.../server/ThrottledChannelExpirationTest.scala | 21 +------
.../kafka/server/quota/ThrottleCallback.java | 22 +++++++
.../kafka/server/quota/ThrottledChannel.java | 72 ++++++++++++++++++++++
.../kafka/server/quota/ThrottledChannelTest.java | 56 +++++++++++++++++
10 files changed, 157 insertions(+), 81 deletions(-)
diff --git a/core/src/main/scala/kafka/server/ClientQuotaManager.scala b/core/src/main/scala/kafka/server/ClientQuotaManager.scala
index a8c6a2fc557..bd12a51d3f3 100644
--- a/core/src/main/scala/kafka/server/ClientQuotaManager.scala
+++ b/core/src/main/scala/kafka/server/ClientQuotaManager.scala
@@ -29,7 +29,7 @@ import org.apache.kafka.common.metrics.stats.{Avg, CumulativeSum, Rate}
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.utils.{Sanitizer, Time}
import org.apache.kafka.server.config.{ClientQuotaManagerConfig, ZooKeeperInternals}
-import org.apache.kafka.server.quota.{ClientQuotaCallback, ClientQuotaEntity, ClientQuotaType}
+import org.apache.kafka.server.quota.{ClientQuotaCallback, ClientQuotaEntity, ClientQuotaType, ThrottleCallback, ThrottledChannel}
import org.apache.kafka.server.util.ShutdownableThread
import org.apache.kafka.network.Session
diff --git a/core/src/main/scala/kafka/server/RequestHandlerHelper.scala b/core/src/main/scala/kafka/server/RequestHandlerHelper.scala
index 1d831f135b2..dbc824021be 100644
--- a/core/src/main/scala/kafka/server/RequestHandlerHelper.scala
+++ b/core/src/main/scala/kafka/server/RequestHandlerHelper.scala
@@ -27,6 +27,7 @@ import org.apache.kafka.common.network.Send
import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse}
import org.apache.kafka.common.utils.Time
import org.apache.kafka.coordinator.group.GroupCoordinator
+import org.apache.kafka.server.quota.ThrottleCallback
import java.util.OptionalInt
diff --git a/core/src/main/scala/kafka/server/ThrottledChannel.scala b/core/src/main/scala/kafka/server/ThrottledChannel.scala
deleted file mode 100644
index 809167857e5..00000000000
--- a/core/src/main/scala/kafka/server/ThrottledChannel.scala
+++ /dev/null
@@ -1,61 +0,0 @@
-/**
- * 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 kafka.server
-
-import java.util.concurrent.{Delayed, TimeUnit}
-
-import kafka.utils.Logging
-import org.apache.kafka.common.utils.Time
-
-trait ThrottleCallback {
- def startThrottling(): Unit
- def endThrottling(): Unit
-}
-
-/**
- * Represents a request whose response has been delayed.
- * @param time Time instance to use
- * @param throttleTimeMs Delay associated with this request
- * @param callback Callback for channel throttling
- */
-class ThrottledChannel(
- val time: Time,
- val throttleTimeMs: Int,
- val callback: ThrottleCallback
-) extends Delayed with Logging {
-
- private val endTimeNanos = time.nanoseconds() + TimeUnit.MILLISECONDS.toNanos(throttleTimeMs)
-
- // Notify the socket server that throttling has started for this channel.
- callback.startThrottling()
-
- // Notify the socket server that throttling has been done for this channel.
- def notifyThrottlingDone(): Unit = {
- trace(s"Channel throttled for: $throttleTimeMs ms")
- callback.endThrottling()
- }
-
- override def getDelay(unit: TimeUnit): Long = {
- unit.convert(endTimeNanos - time.nanoseconds(), TimeUnit.NANOSECONDS)
- }
-
- override def compareTo(d: Delayed): Int = {
- val other = d.asInstanceOf[ThrottledChannel]
- java.lang.Long.compare(this.endTimeNanos, other.endTimeNanos)
- }
-}
diff --git a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
index d853a4e212d..8a7e0695605 100644
--- a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
+++ b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
@@ -40,6 +40,7 @@ import org.apache.kafka.security.CredentialProvider
import org.apache.kafka.server.common.{FinalizedFeatures, MetadataVersion}
import org.apache.kafka.server.config.QuotaConfigs
import org.apache.kafka.server.metrics.KafkaYammerMetrics
+import org.apache.kafka.server.quota.{ThrottleCallback, ThrottledChannel}
import org.apache.kafka.test.{TestSslUtils, TestUtils => JTestUtils}
import org.apache.log4j.Level
import org.junit.jupiter.api.Assertions._
diff --git a/core/src/test/scala/unit/kafka/server/BaseClientQuotaManagerTest.scala b/core/src/test/scala/unit/kafka/server/BaseClientQuotaManagerTest.scala
index af4e08f5586..ad13fa3a4cd 100644
--- a/core/src/test/scala/unit/kafka/server/BaseClientQuotaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/BaseClientQuotaManagerTest.scala
@@ -30,6 +30,7 @@ import org.apache.kafka.common.requests.{AbstractRequest, FetchRequest, RequestC
import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
import org.apache.kafka.common.utils.MockTime
import org.apache.kafka.network.Session
+import org.apache.kafka.server.quota.ThrottleCallback
import org.junit.jupiter.api.AfterEach
import org.mockito.Mockito.mock
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
index 62d12a87e28..cebe1b28836 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
@@ -87,6 +87,7 @@ import org.apache.kafka.server.common.{FinalizedFeatures, KRaftVersion, Metadata
import org.apache.kafka.server.config.{ConfigType, KRaftConfigs, ReplicationConfigs, ServerConfigs, ServerLogConfigs, ShareGroupConfig}
import org.apache.kafka.server.metrics.ClientMetricsTestUtils
import org.apache.kafka.server.share.{CachedSharePartition, ErroneousAndValidPartitionData, FinalContext, ShareAcknowledgementBatch, ShareSession, ShareSessionContext, ShareSessionKey}
+import org.apache.kafka.server.quota.ThrottleCallback
import org.apache.kafka.server.util.{FutureUtils, MockTime}
import org.apache.kafka.storage.internals.log.{AppendOrigin, FetchParams, FetchPartitionData, LogConfig}
import org.junit.jupiter.api.Assertions._
diff --git a/core/src/test/scala/unit/kafka/server/ThrottledChannelExpirationTest.scala b/core/src/test/scala/unit/kafka/server/ThrottledChannelExpirationTest.scala
index 3d3d5732247..06def9021ce 100644
--- a/core/src/test/scala/unit/kafka/server/ThrottledChannelExpirationTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ThrottledChannelExpirationTest.scala
@@ -19,10 +19,11 @@ package kafka.server
import java.util.Collections
-import java.util.concurrent.{DelayQueue, TimeUnit}
+import java.util.concurrent.DelayQueue
import org.apache.kafka.common.metrics.MetricConfig
import org.apache.kafka.common.utils.MockTime
import org.apache.kafka.server.config.ClientQuotaManagerConfig
+import org.apache.kafka.server.quota.{ThrottleCallback, ThrottledChannel}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{BeforeEach, Test}
@@ -81,22 +82,4 @@ class ThrottledChannelExpirationTest {
clientMetrics.shutdown()
}
}
-
- @Test
- def testThrottledChannelDelay(): Unit = {
- val t1: ThrottledChannel = new ThrottledChannel(time, 10, callback)
- val t2: ThrottledChannel = new ThrottledChannel(time, 20, callback)
- val t3: ThrottledChannel = new ThrottledChannel(time, 20, callback)
- assertEquals(10, t1.throttleTimeMs)
- assertEquals(20, t2.throttleTimeMs)
- assertEquals(20, t3.throttleTimeMs)
-
- for (itr <- 0 to 2) {
- assertEquals(10 - 10*itr, t1.getDelay(TimeUnit.MILLISECONDS))
- assertEquals(20 - 10*itr, t2.getDelay(TimeUnit.MILLISECONDS))
- assertEquals(20 - 10*itr, t3.getDelay(TimeUnit.MILLISECONDS))
- time.sleep(10)
- }
- }
-
}
diff --git a/server/src/main/java/org/apache/kafka/server/quota/ThrottleCallback.java b/server/src/main/java/org/apache/kafka/server/quota/ThrottleCallback.java
new file mode 100644
index 00000000000..e88be6be6e2
--- /dev/null
+++ b/server/src/main/java/org/apache/kafka/server/quota/ThrottleCallback.java
@@ -0,0 +1,22 @@
+/*
+ * 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.server.quota;
+
+public interface ThrottleCallback {
+ void startThrottling();
+ void endThrottling();
+}
diff --git a/server/src/main/java/org/apache/kafka/server/quota/ThrottledChannel.java b/server/src/main/java/org/apache/kafka/server/quota/ThrottledChannel.java
new file mode 100644
index 00000000000..73f4208936f
--- /dev/null
+++ b/server/src/main/java/org/apache/kafka/server/quota/ThrottledChannel.java
@@ -0,0 +1,72 @@
+/*
+ * 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.server.quota;
+
+import org.apache.kafka.common.utils.Time;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Delayed;
+import java.util.concurrent.TimeUnit;
+
+public class ThrottledChannel implements Delayed {
+ private static final Logger LOGGER = LoggerFactory.getLogger(ThrottledChannel.class);
+ private final Time time;
+ private final int throttleTimeMs;
+ private final ThrottleCallback callback;
+ private final long endTimeNanos;
+
+ /**
+ * Represents a request whose response has been delayed.
+ * @param time Time instance to use
+ * @param throttleTimeMs Delay associated with this request
+ * @param callback Callback for channel throttling
+ */
+ public ThrottledChannel(Time time, int throttleTimeMs, ThrottleCallback callback) {
+ this.time = time;
+ this.throttleTimeMs = throttleTimeMs;
+ this.callback = callback;
+ this.endTimeNanos = time.nanoseconds() + TimeUnit.MILLISECONDS.toNanos(throttleTimeMs);
+
+ // Notify the socket server that throttling has started for this channel.
+ callback.startThrottling();
+ }
+
+ /**
+ * Notify the socket server that throttling has been done for this channel.
+ */
+ public void notifyThrottlingDone() {
+ LOGGER.trace("Channel throttled for: {} ms", throttleTimeMs);
+ callback.endThrottling();
+ }
+
+ @Override
+ public long getDelay(TimeUnit unit) {
+ return unit.convert(endTimeNanos - time.nanoseconds(), TimeUnit.NANOSECONDS);
+ }
+
+ @Override
+ public int compareTo(Delayed other) {
+ ThrottledChannel otherChannel = (ThrottledChannel) other;
+ return Long.compare(this.endTimeNanos, otherChannel.endTimeNanos);
+ }
+
+ public int throttleTimeMs() {
+ return throttleTimeMs;
+ }
+}
diff --git a/server/src/test/java/org/apache/kafka/server/quota/ThrottledChannelTest.java b/server/src/test/java/org/apache/kafka/server/quota/ThrottledChannelTest.java
new file mode 100644
index 00000000000..64f37bdc378
--- /dev/null
+++ b/server/src/test/java/org/apache/kafka/server/quota/ThrottledChannelTest.java
@@ -0,0 +1,56 @@
+/*
+ * 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.server.quota;
+
+import org.apache.kafka.common.utils.MockTime;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ThrottledChannelTest {
+
+ private final MockTime time = new MockTime();
+ private final ThrottleCallback callback = new ThrottleCallback() {
+ @Override
+ public void startThrottling() {
+ }
+
+ @Override
+ public void endThrottling() {
+ }
+ };
+
+ @Test
+ public void testThrottledChannelDelay() {
+ ThrottledChannel channel1 = new ThrottledChannel(time, 10, callback);
+ ThrottledChannel channel2 = new ThrottledChannel(time, 20, callback);
+ ThrottledChannel channel3 = new ThrottledChannel(time, 20, callback);
+ assertEquals(10, channel1.throttleTimeMs());
+ assertEquals(20, channel2.throttleTimeMs());
+ assertEquals(20, channel3.throttleTimeMs());
+
+ for (int i = 0; i <= 2; i++) {
+ assertEquals(10 - 10 * i, channel1.getDelay(TimeUnit.MILLISECONDS));
+ assertEquals(20 - 10 * i, channel2.getDelay(TimeUnit.MILLISECONDS));
+ assertEquals(20 - 10 * i, channel3.getDelay(TimeUnit.MILLISECONDS));
+ time.sleep(10);
+ }
+ }
+}