From a22894df3644f8df21460c93d7b5da58d4f969f0 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Gell=C3=A9rt=20Peresztegi-Nagy?= <gellert.nagy@redpanda.com>
Date: Mon, 24 Jun 2024 21:51:36 +0100
Subject: [PATCH] k/quotas: shortcut quota methods if no quotas

Minimize the impact of client quota management in the default (and
expectly most common case) of having no quotas configured.
---
 src/v/kafka/server/client_quota_translator.cc        |  9 +++++++++
 src/v/kafka/server/client_quota_translator.h         |  3 +++
 src/v/kafka/server/quota_manager.cc                  | 12 ++++++++++++
 .../server/tests/client_quota_translator_test.cc     |  2 ++
 4 files changed, 26 insertions(+)

diff --git a/src/v/kafka/server/client_quota_translator.cc b/src/v/kafka/server/client_quota_translator.cc
index cdc0d08ab8bc8..3bf0f9604f871 100644
--- a/src/v/kafka/server/client_quota_translator.cc
+++ b/src/v/kafka/server/client_quota_translator.cc
@@ -16,6 +16,8 @@
 
 #include <seastar/util/variant_utils.hh>
 
+#include <absl/algorithm/container.h>
+
 #include <optional>
 
 namespace kafka {
@@ -288,4 +290,11 @@ client_quota_translator::get_default_config(client_quota_type qt) const {
     }
 }
 
+bool client_quota_translator::is_empty() const {
+    return _quota_store.local().size() == 0
+           && absl::c_all_of(all_client_quota_types, [this](auto qt) {
+                  return !get_default_config(qt);
+              });
+}
+
 } // namespace kafka
diff --git a/src/v/kafka/server/client_quota_translator.h b/src/v/kafka/server/client_quota_translator.h
index c0c54c9fa0489..4773935e6bde1 100644
--- a/src/v/kafka/server/client_quota_translator.h
+++ b/src/v/kafka/server/client_quota_translator.h
@@ -141,6 +141,9 @@ class client_quota_translator {
     /// `watch` can be used to register for quota changes
     void watch(on_change_fn&& fn);
 
+    /// Returns true if there are no quotas configured
+    bool is_empty() const;
+
 private:
     using quota_config
       = std::unordered_map<ss::sstring, config::client_group_quota>;
diff --git a/src/v/kafka/server/quota_manager.cc b/src/v/kafka/server/quota_manager.cc
index 96a135bf1fd6e..085a32d483b32 100644
--- a/src/v/kafka/server/quota_manager.cc
+++ b/src/v/kafka/server/quota_manager.cc
@@ -320,6 +320,9 @@ ss::future<std::chrono::milliseconds> quota_manager::record_partition_mutations(
     /// KIP-599 throttles create_topics / delete_topics / create_partitions
     /// request. This delay should only be applied to these requests if the
     /// quota has been exceeded
+    if (_translator.is_empty()) {
+        co_return 0ms;
+    }
     auto ctx = client_quota_request_ctx{
       .q_type = client_quota_type::partition_mutation_quota,
       .client_id = client_id,
@@ -391,6 +394,9 @@ ss::future<clock::duration> quota_manager::record_produce_tp_and_throttle(
   std::optional<std::string_view> client_id,
   uint64_t bytes,
   clock::time_point now) {
+    if (_translator.is_empty()) {
+        co_return 0ms;
+    }
     auto ctx = client_quota_request_ctx{
       .q_type = client_quota_type::produce_quota,
       .client_id = client_id,
@@ -440,6 +446,9 @@ ss::future<> quota_manager::record_fetch_tp(
   std::optional<std::string_view> client_id,
   uint64_t bytes,
   clock::time_point now) {
+    if (_translator.is_empty()) {
+        co_return;
+    }
     auto ctx = client_quota_request_ctx{
       .q_type = client_quota_type::fetch_quota,
       .client_id = client_id,
@@ -469,6 +478,9 @@ ss::future<> quota_manager::record_fetch_tp(
 
 ss::future<clock::duration> quota_manager::throttle_fetch_tp(
   std::optional<std::string_view> client_id, clock::time_point now) {
+    if (_translator.is_empty()) {
+        co_return 0ms;
+    }
     auto ctx = client_quota_request_ctx{
       .q_type = client_quota_type::fetch_quota,
       .client_id = client_id,
diff --git a/src/v/kafka/server/tests/client_quota_translator_test.cc b/src/v/kafka/server/tests/client_quota_translator_test.cc
index 231cabecb7f62..682d9a6248942 100644
--- a/src/v/kafka/server/tests/client_quota_translator_test.cc
+++ b/src/v/kafka/server/tests/client_quota_translator_test.cc
@@ -96,6 +96,7 @@ SEASTAR_THREAD_TEST_CASE(quota_translator_default_test) {
     auto limits = f.tr.find_quota_value(key);
     BOOST_CHECK_EQUAL(test_client_id_key, key);
     BOOST_CHECK_EQUAL(default_limits, limits);
+    BOOST_CHECK(f.tr.is_empty());
 }
 
 SEASTAR_THREAD_TEST_CASE(quota_translator_modified_default_test) {
@@ -116,6 +117,7 @@ SEASTAR_THREAD_TEST_CASE(quota_translator_modified_default_test) {
     auto limits = f.tr.find_quota_value(key);
     BOOST_CHECK_EQUAL(test_client_id_key, key);
     BOOST_CHECK_EQUAL(expected_limits, limits);
+    BOOST_CHECK(!f.tr.is_empty());
 }
 
 void run_quota_translator_client_group_test(fixture& f) {