...

This KIP introduces a new interface, ClientQuotaCallbackHandler, which adds a new method method updateClusterMetadata(ClusterMetadata). The to replace the old one, the parameter ClusterMetadata in the new method is a new interface, which will be described below. This new interface will include all existing methods from ClientQuotaCallback, except for the deprecated updateClusterMetadata(Cluster) method.Normally, to deprecate a method in an interface, we can simply add a new method to the existing interface. However, the proposal here aims to eliminate the creation of Cluster objects, so we will use an alternative approach: introduce a new interface. This will allow us to use instanceof to determine whether we should use the old or the more efficient method to trigger updateClusterMetadata

For backward compatibility, the new method updateClusterMetadata(ClusterMetadata) have a default implementation and trigger the old one.

@@ -101,13 +114,82 @@ public interface ClientQuotaCallbackHandlerClientQuotaCallback extends Configurable {
    // All methods except updateClusterMetadata(Cluster) in ClientQuotaCallback are now placed here.

    /**
     * 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 please use {@link ClientQuotaCallback#updateClusterMetadata(ClusterMetadata)} instead
      * @param clusterMetadatacluster Cluster metadata including partitions and their leaders if known
      * @return true if quotas have changed and metric configs may need to be updated
      */
+    default boolean updateClusterMetadata(ClusterMetadata clusterMetadata) {@Deprecated
     boolean   throw new UnsupportedOperationException(updateClusterMetadata(Cluster cluster);
    }
}

The ClientQuotaCallback interface now extends the new interface ClientQuotaCallbackHandler. It is deprecated and expected to be removed in future major releases.

/**
 * Deprecated, please use ClientQuotaCallbackHandler instead
 */
@Deprecated
public interface ClientQuotaCallback extends ClientQuotaCallbackHandler {


+    /**
+     * 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 please use {@link ClientQuotaCallback#updateClusterMetadata(ClusterMetadata)} instead
     * @param clusterclusterMetadata Cluster metadata including partitions and their leaders, if known
+     * @return true if quotas have changed and metric configs may need to be updated
+     */
+    default boolean updateClusterMetadata(ClusterMetadata clusterMetadata) @Deprecated{
+    boolean    return updateClusterMetadata(toCluster(Cluster cluster);
clusterMetadata));
+    }
+
     /**
      * Closes this instance.
      */
     void close();
+
+    static Cluster toCluster(ClusterMetadata clusterMetadata) {
+        // ... skip the implementation here ...
+    }

The ClusterMetadata used in ClientQuotaCallbackHandler#updateClusterMetadata ClientQuotaCallback#updateClusterMetadata is also a new interface. The goal of ClusterMetadata is:

...

public interface ClusterMetadata {

    /**
     * Get a map of brokers.
     * The key is the broker node ID, and the value is the metadata
     * associated with that broker.
     */
    Map<Integer, BrokerMetadata> brokers();

    /**
     * Get specific broker information.
     */
    Optional<BrokerMetadata> broker(int nodeId);

    /**
     * Get partition information for a specific topic.
     * The key is the partition ID, and the value is the metadata
     * associated with that partition.
     * Return an empty map if the topic does not exist.
     */
    Map<Integer, PartitionMetadata> partitionsForTopic(String topic);

    /**
     * Get partition information for a specific node.
     * The key is the TopicPartition, and the value is the metadata
     * associated with that partition.
     * Return an empty map if the node does not exist.
     */
    Map<TopicPartition, PartitionMetadata> partitionsForNode(int nodeId);

    /**
     * Get specific partition information.
     */
    Optional<PartitionMetadata> partition(TopicPartition topicPartition);

    /**
     * Get a map of topics.
     * The key is the topic ID, and the value is the topic name.
     */
    Map<Uuid, String> topics();

    /**
     * Get ClusterResource, which includes the cluster ID.
     */
    ClusterResource clusterResource();

    /**
     * Given partition metadata, return the subset of the replicas that are offline.
     * If there is no offline replica in the partition, return an empty array.
     */
    int[] offlineReplicas(PartitionMetadata partitionMetadata);
}

...

Unlike the Cluster#brokers using the org.apache.kafka.common.Node to present broker information.  ClusterMetadata use the new BrokerMetadata interface. The reason behind it is  ClientQuotaCallback#updateClusterMetadata does not filter by listener name like the old ZK-based path. Thus, using the existing Node is is not appropriate since now we may have multiple listeners in the broker node. Additionally, DynamicTopicClusterQuotaPublisher can access BrokerRegistration, and we can simply add a new interface that to BrokerRegistration which already includes the necessary information in BrokerRegistration, making the ClusterMetadata implementation much more efficient.

public interface BrokerMetadata {

    /**
     * The broker node IDid
     */
    int id();

    /**
     * The listeners of the broker node
     */
    Map<String, Endpoint> listeners();

    /**
     * Whether if this node is fenced
     */
    boolean fenced();

    /**
     * The rack for this node
     */
    Optional<String> rack();
}

PartitionMetadata is similar to BrokerMetadata. We can leverage the existing PartitionRegistration object to create a new interface that provides partition metadata efficiently, avoiding the need to convert PartitionRegistration into PartitionInfo.

...

In DynamicTopicClusterQuotaPublisher#onMetadataUpdate,   determine which interface the user has implemented and decide which updateClusterMetadata method to callinvoke the new method introduced in this KIP.

clientQuotaCallback match@@ -52,8 +51,8 @@ class DynamicTopicClusterQuotaPublisher (
     try {
     case _: ClientQuotaCallback =>
 quotaManagers.clientQuotaCallbackPlugin().ifPresent(plugin => {
         if (delta.topicsDelta() != null || delta.clusterDelta() != null) {
-          val cluster = MetadataCache.toCluster(clusterId, newImage)
-          if (plugin.get(clientQuotaCallback).updateClusterMetadata(cluster)) {
  +    ...
    }
  case _: ClientQuotaCallbackHandler =>
    val clusterMetadata = new DefaultClusterMetadata(clusterId, newImage)
+          if (plugin.get(clientQuotaCallback).updateClusterMetadata(clusterMetadata)) {
             quotaManagers.fetch..updateQuotaMetricConfigs()
    }         quotaManagers.produce.updateQuotaMetricConfigs()
  case _ =>
    throw new IllegalArgumentException("Unsupported client quota callback")
} quotaManagers.request.updateQuotaMetricConfigs()

Make PartitionRegistration implement PartitionMetadata and BrokerRegistration implement BrokerMetadata. In DefaultClusterMetadata (the implementation of ClusterMetadata), we can use BrokerRegistration and PartitionRegistration to return the necessary information without duplicating data. This avoids the overhead of creating new objects.

...

public Map<Integer, PartitionMetadata> partitionsForTopic(String topic) {
    return imagemetadataImage.topics().getTopic(topic).partitions().entrySet().stream()
            .collect(Collectors.toMaptoUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
}

The implementation for ClusterMetadata#partition(TopicPartition topicPartition) is similarly efficient.

public PartitionMetadata partition(TopicPartition topicPartition) {
    return imagemetadataImage.topics().getTopic(topicPartition.topic()).partitions().get(topicPartition.partition());
}

Other methods in ClusterMetadata will follow a similar pattern.

...

This KIP should be backward-compatible, as the deprecated interface ClientQuotaCallback is method updateClusterMetadata(Cluster) is still supported and invoked in the new method.

However, users are expected to implement the new interface ClientQuotaCallbackHandler. The ClientQuotaCallback interface will method ClientQuotaCallback#updateClusterMetadata(ClusterMetadata). The old one - ClientQuotaCallback#updateClusterMetadata(Cluster) will be removed in a future major release.

...

We will rely on the existing CustomQuotaCallbackTest.java to ensure the deprecated interface method still works.

Additionally, we will add

...

some new tests in this test class to

...

ensure the new

...

method work as expected.

Rejected Alternatives

Add new replacement method to the existing interface - ClientQuotaCallback.java

The proposal here aims to eliminate the creation of Cluster objects, so the following pseudo code below cannot meet the requirement.

We will use an alternative approach: introduce a new interface thus we can decide which interface user is used and then trigger the corresponding updateClusterMetadata.

...

...

N/A