KAFKA-19566: Deprecate ClientQuotaCallback#updateClusterMetadata

clients/clients-integration-tests/src/test/java/org/apache/kafka/server/quota/CustomQuotaCallbackTest.java

@@ -149,6 +149,7 @@ public boolean quotaResetRequired(ClientQuotaType quotaType) {
             return true;
         }
 
+        @SuppressWarnings("removal")
         @Override
         public boolean updateClusterMetadata(Cluster cluster) {
             COUNTERS.computeIfAbsent(nodeId, k -> new AtomicInteger()).incrementAndGet();

clients/src/main/java/org/apache/kafka/server/quota/ClientQuotaCallback.java

@@ -96,15 +96,19 @@ public interface ClientQuotaCallback extends Configurable {
     boolean quotaResetRequired(ClientQuotaType quotaType);
 
     /**
-     * This callback is invoked whenever there are changes in the cluster metadata, such as 
+     * This callback is invoked whenever there are changes in the cluster metadata, such as
      * brokers being added or removed, topics being created or deleted, or partition leadership updates.
      * This is useful if quota computation takes partitions into account.
      * Topics that are being deleted will not be included in `cluster`.
      *
+     * @deprecated since 4.4 and should not be used any longer.
      * @param cluster Cluster metadata including partitions and their leaders if known
      * @return true if quotas have changed and metric configs may need to be updated
      */
-    boolean updateClusterMetadata(Cluster cluster);
+    @Deprecated(since = "4.4", forRemoval = true)
+    default boolean updateClusterMetadata(Cluster cluster) {
+        return false;
+    }
 
     /**
      * Closes this instance.

core/src/test/scala/integration/kafka/api/CustomQuotaCallbackTest.scala

@@ -452,6 +452,7 @@ class GroupedUserQuotaCallback extends ClientQuotaCallback with Reconfigurable w
     if (group != null) quotaOrDefault(group, quotaType) else null
   }
 
+  @SuppressWarnings(Array("removal"))
   override def updateClusterMetadata(cluster: Cluster): Boolean = {
     val topicsByGroup = cluster.topics.asScala.groupBy(group)
 

core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala

@@ -16,7 +16,6 @@
  */
 package kafka.server
 
-import org.apache.kafka.common.Cluster
 import java.net.InetAddress
 import org.apache.kafka.common.internals.Plugin
 import org.apache.kafka.common.metrics.Quota
@@ -571,7 +570,6 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
       override def quotaMetricTags(quotaType: ClientQuotaType, principal: KafkaPrincipal, clientId: String): Map[String, String] = Collections.emptyMap()
 
       override def quotaLimit(quotaType: ClientQuotaType, metricTags: Map[String, String]): java.lang.Double = 1
-      override def updateClusterMetadata(cluster: Cluster): Boolean = false
 
       override def updateQuota(quotaType: ClientQuotaType, entity: ClientQuotaEntity, newValue: Double): Unit = {
         quotas.put(entity.asInstanceOf[ClientQuotaManager.KafkaQuotaEntity], new Quota(newValue.toLong, true))

docs/getting-started/upgrade.md

@@ -26,6 +26,14 @@ type: docs
 -->
 
 
+## Upgrading to 4.4.0
+
+### Upgrading Servers to 4.4.0 from any version 3.3.x through 4.3.0
+
+### Notable changes in 4.4.0
+
+  * The `ClientQuotaCallback#updateClusterMetadata` method is deprecated and will be removed in Kafka 5.0. Custom implementations of `ClientQuotaCallback` no longer need to override this method, as a default no-op implementation is now provided. For further details, please refer to [KIP-1200](https://cwiki.apache.org/confluence/x/axBJFg).
+
 ## Upgrading to 4.3.0
 
 ### Upgrading Servers to 4.3.0 from any version 3.3.x through 4.2.0

metadata/src/main/java/org/apache/kafka/metadata/publisher/DynamicTopicClusterQuotaPublisher.java

@@ -57,6 +57,7 @@ public String name() {
         return "DynamicTopicClusterQuotaPublisher " + nodeType + " id=" + nodeId;
     }
 
+    @SuppressWarnings("removal")
     @Override
     public void onMetadataUpdate(MetadataDelta delta, MetadataImage newImage, LoaderManifest manifest) {
         try {

server/src/main/java/org/apache/kafka/server/quota/ClientQuotaManager.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.server.quota;
 
-import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.internals.Plugin;
 import org.apache.kafka.common.metrics.MetricConfig;
@@ -796,12 +795,6 @@ private Quota findClientQuota(ClientIdEntity clientIdEntity) {
             return overriddenQuotas.get(DEFAULT_CLIENT_ID_QUOTA_ENTITY);
         }
 
-        @Override
-        public boolean updateClusterMetadata(Cluster cluster) {
-            // The default quota callback does not use any cluster metadata
-            return false;
-        }
-
         @Override
         public void updateQuota(ClientQuotaType quotaType, ClientQuotaEntity entity, double newValue) {
             KafkaQuotaEntity quotaEntity = (KafkaQuotaEntity) entity;

server/src/test/java/org/apache/kafka/server/KRaftClusterTest.java

@@ -34,7 +34,6 @@
 import org.apache.kafka.clients.admin.QuorumInfo;
 import org.apache.kafka.clients.admin.SupportedVersionRange;
 import org.apache.kafka.clients.admin.TopicDescription;
-import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.Endpoint;
 import org.apache.kafka.common.KafkaFuture;
 import org.apache.kafka.common.Node;
@@ -1779,11 +1778,6 @@ public boolean quotaResetRequired(ClientQuotaType quotaType) {
             return true;
         }
 
-        @Override
-        public boolean updateClusterMetadata(Cluster cluster) {
-            return false;
-        }
-
         @Override
         public void close() {
         }