...

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

In ClientQuotaCallback#updateClusterMetadata, we pass a Cluster object to this method. The Cluster is an immutable object that holds a lot of information, and a new Cluster is created every time there is a change in metadata. For large clusters with many partitions, this can result in significant memory pressure on Kafka.

Additionally, some information in the Cluster object is unnecessary or confusing. For instance, Cluster#controller refers to a random broker node in a KRaft cluster, which is retained for backward compatibility as it represents the ZK controller. Furthermore, in ZK mode, we parse the listener using the listener name from the request. However, in KRaft mode, there is no listener name in the updateClusterMetadata path. As a result, the callback may receive multiple partition info entries for each listener name. (See KAFKA-19122).

To address these issues, this KIP propose deprecating the current method ClientQuotaCallback#updateClusterMetadata(Cluster) and reimplementing it in a more efficient way, providing clearer and more relevant cluster information.

Public Interfaces

This KIP introduces a new

...

interface ClientQuotaCallbackHandler which basically copy all methods in ClientQuotaCallback except the one we want to deprecate.

boolean updateClusterMetadata(Cluster cluster);

And add a new method to replace the old one, the parameter ClusterMetadata in the new method is a new interface, which will be described below.For backward compatibility, the new method

boolean updateClusterMetadata(ClusterMetadata clusterMetadata);

Here is the whole content of the new interface:) have a default implementation and trigger the old one.

@@ -101,13 +114,82 @@ public interface ClientQuotaCallback extendsClientQuotaCallbackHandler Configurable {

    /**
  * This is useful* ifQuota quotacallback computation takes partitions into accountinvoked to determine the quota metric tags to be applied for a request.
      * TopicsQuota thatlimits are beingassociated deletedwith willquota notmetrics beand includedall inclients `cluster`.
which use the    *same
+     * @deprecatedmetric pleasetags useshare {@link ClientQuotaCallback#updateClusterMetadata(ClusterMetadata)} instead
 the quota limit.
     *
 @param cluster Cluster metadata including* partitions@param andquotaType theirType leadersof ifquota knownrequested
      * @return@param trueprincipal ifThe quotasuser haveprincipal changedof andthe metricconnection configsfor maywhich needquota tois be updatedrequested
     * */
+@param clientId  The @Deprecated
client id associated with the booleanrequest
 updateClusterMetadata(Cluster cluster);

+    /**
+ @return quota metric tags *that Thisindicate callbackwhich isother invokedclients whenevershare therethis arequota
 changes in the cluster metadata,*/
 such as
+  Map<String, String> quotaMetricTags(ClientQuotaType *quotaType, brokers being added or removedKafkaPrincipal principal, topics being created or deleted, or partition leadership updates.
+String clientId);

    /**
     * ThisReturns isthe usefulquota iflimit quotaassociated computationwith takesthe partitionsprovided intometric accounttags.
+ These tags were returned *from
 Topics that are being deleted* willa notprevious becall includedto in `cluster`.
+     *
+{@link #quotaMetricTags(ClientQuotaType, KafkaPrincipal, String)}. This method is
     * @paraminvoked clusterMetadataby Clusterquota metadatamanagers includingto partitionsobtain andthe theircurrent leadersquota iflimit known
+applied to a metric when *the @returnfirst truerequest
 if quotas have changed and* metricusing configsthese maytags needis toprocessed. beIt updated
+is also invoked after a */
+quota update or cluster defaultmetadata boolean updateClusterMetadata(ClusterMetadata clusterMetadata) {
+        return updateClusterMetadata(toCluster(clusterMetadata));
+    }
+
     /**
      * Closes this instance.
      */
     void close();
+
+    static Cluster toCluster(ClusterMetadata clusterMetadata) {
+        // ... skip the implementation here ...
+    }

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

• Eliminate the Cluster object creation, allowing us to avoid the overhead of creating Cluster instances.

• Provide key information from the Cluster object, such as brokers and partition data, along with helper methods that replicate the functionality of the original Cluster class while omitting unnecessary information (e.g., controller(), unauthorizedTopics(), isBootstrapConfigured()).

public interface ClusterMetadata {      
    
    change.
     * If the tags are no longer in use after the update, (e.g. this is a {user, client-id} quota metric
     * and the quota now in use is a {user} quota), null is returned.
     *
     * @param quotaType  Type of quota requested
     * @param metricTags Metric tags for a quota metric of type `quotaType`
     * @return the quota limit for the provided metric tags or null if the metric tags are no longer in use
     */
    Double quotaLimit(ClientQuotaType quotaType, Map<String, String> metricTags);

    /**
     * GetQuota aconfiguration listupdate ofcallback Kafkathat brokeris nodeinvoked IDswhen inquota theconfiguration cluster.
for an entity   */is
    List<Integer> brokerIds();

    /**
     * Get specific broker information. * updated in the quorum. This is useful to track configured quotas if built-in quota configuration
     */
 tools are used Optional<BrokerMetadata>for broker(int nodeId);

quota management.
     /**
     * Get@param specificquotaType partition information.
 Type of quota being */updated
    PartitionMetadata partition(TopicPartition topicPartition);

    /**
 @param quotaEntity The quota *entity Getfor partitionwhich informationquota ofis specificbeing topic.updated
     * The@param keynewValue is the partition id,The andnew thequota value is the metadata
     */
 associated with that partition
void updateQuota(ClientQuotaType quotaType,   */
    Map<IntegerClientQuotaEntity quotaEntity, PartitionMetadata> partitionsForTopic(String topicdouble newValue);

    /**
     * GetQuota partitionconfiguration informationremoval ofcallback specificthat node.
is invoked when quota configuration *for Thean keyentity is
 the TopicPartition, and the value* isremoved in the metadata
     * associated with that partition. quorum. This is useful to track configured quotas if built-in quota configuration
     */
 tools are used Map<TopicPartition,for PartitionMetadata> partitionsForNode(int nodeId);

quota management.
     /**
     * Get@param aquotaType map of topics.
Type of quota being updated
     * The@param keyquotaEntity isThe thequota topicentity id,for andwhich thequota value is the topic name.being updated
     */
    void removeQuota(ClientQuotaType Map<UuidquotaType, String>ClientQuotaEntity topics(quotaEntity);

    /**
     * GetReturns ClusterResourcetrue thatif includesany clusterof id
the existing quota configs may */
have been updated since ClusterResource clusterResource();

    /**the last call
     * Givento athis partition metadata, returnmethod for the subsetprovided ofquota thetype. replicasQuota thatupdates areas offline
a result of calls  */to
    List<Integer> 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 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 to BrokerRegistration which already includes the necessary information, making the ClusterMetadata implementation much more efficient.

public interface BrokerMetadata {

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

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

    /** * {@link #updateClusterMetadata(ClusterMetadata)}, {@link #updateQuota(ClientQuotaType, ClientQuotaEntity, double)}
     * and {@link #removeQuota(ClientQuotaType, ClientQuotaEntity)} are automatically processed.
     * So callbacks that rely only on built-in quota configuration tools always return false. Quota callbacks
     * with external quota configuration or custom reconfigurable quota configs that affect quota limits must
     * return true if existing metric configs may need to be updated. This method is invoked on every request
     * and hence is expected to be handled by callbacks as a simple flag that is updated when quotas change.
     *
     * Whether@param ifquotaType thisType nodeof is fencedquota
     */
    boolean fencedquotaResetRequired(ClientQuotaType quotaType);

    /**
     * TheThis rackcallback foris thisinvoked node
whenever there are changes in the cluster metadata, such as
  */
    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.

The main difference is isAvailable(), as PartitionRegistration does not include this information. However, since Cluster#availablePartitionsForTopic allows users to query available partitions, providing a default method here for convenience.

public interface PartitionMetadata {

    /**   * 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.
     *
 The node id of the* node@param currentlyclusterMetadata actingCluster asmetadata aincluding leaderpartitions forand thistheir partitionleaders or -1 if there is no leaderknown
     */
 @return true if int leader();

    /**
     * The complete set of replicas for this partition regardless of whether they are alive or up-to-datequotas have changed and metric configs may need to be updated
     */
    List<Integer>boolean replicasupdateClusterMetadata(ClusterMetadata clusterMetadata);

    /**
     * TheCloses subsetthis ofinstance.
 the replicas that are in*/
 sync, that is void close();
}

For backward compatibility, the old interface ClientQuotaCallback will extend the new interface ClientQuotaCallbackHandler, and in the old interface we'll add a new method updateClusterMetadata(ClusterMetadata) have a default implementation and trigger the old one.

The toCluster method is a static method to covert the ClusterMetadata object to Cluster object.

@Deprecated
public interface ClientQuotaCallback extends ClientQuotaCallbackHandler {

    /**
     * This callback is invoked whenever there are changes in the cluster metadata, such as caught-up to the leader and ready to take over as leader if
     * the leader should fail. Note that there is no guarantee the first element is leader, please use leader()
     * to get the leader of partition.
     */
 brokers being added List<Integer> inSyncReplicas();

    /**
     * Check if the partition is available
     */
    default boolean isAvailable() {
       return leader() != Node.noNode().id();
    }
}

Proposed Changes

In DynamicTopicClusterQuotaPublisher#onMetadataUpdate, invoke the new method introduced in this KIP.

@@ -52,8 +51,8 @@ class DynamicTopicClusterQuotaPublisher (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 ClientQuotaCallbackHandler#updateClusterMetadata(ClusterMetadata)} instead
     try* {
@param cluster Cluster metadata including partitions and  quotaManagers.clientQuotaCallbackPlugin().ifPresent(plugin => {their leaders if known
     * @return true if if (delta.topicsDelta() != null || delta.clusterDelta() != null) {
-quotas have changed and metric configs may need to be updated
     */
    @Deprecated
  val cluster =boolean MetadataCache.toCluster(clusterId, newImage)
-updateClusterMetadata(Cluster cluster);

    default boolean updateClusterMetadata(ClusterMetadata clusterMetadata) {
  if (plugin.get().updateClusterMetadata(cluster)) {
+          val clusterMetadata = new DefaultClusterMetadata(clusterId, newImage)
+          if (plugin.get().updateClusterMetadata(clusterMetadata)) {
             quotaManagers.fetch.updateQuotaMetricConfigs()
             quotaManagers.produce.updateQuotaMetricConfigs()return updateClusterMetadata(toCluster(clusterMetadata));
    }
}

The ClusterMetadata is also a new interface. The goal of ClusterMetadata is:

• Eliminate the Cluster object creation, allowing us to avoid the overhead of creating Cluster instances.

• Provide key information from the Cluster object, such as brokers and partition data, along with helper methods that replicate the functionality of the original Cluster class while omitting unnecessary information (e.g., controller(), unauthorizedTopics(), isBootstrapConfigured()).

public interface ClusterMetadata {      
    
    /**
     * Get a list of Kafka broker node IDs in 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.

For example, the implementation of ClusterMetadata#partitionsForTopic(String) simply wraps the map returned by TopicImage.partitions().

metadataImage.topics().getTopic(topic).partitions().entrySet().stream()
        .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));

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

metadataImage.topics().getTopic(topicPartition.topic()).partitions().get(topicPartition.partition());

Other methods in ClusterMetadata will follow a similar pattern.

Compatibility, Deprecation, and Migration Plan

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

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

Test Plan

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

Additionally, we will add some new tests in this test class to ensure the new method work as expected.

Rejected Alternatives

...

the cluster.
     */
    List<Integer> brokerIds();

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

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

    /**
     * Get partition information of specific topic.
     * The key is the partition id, and the value is the metadata
     * associated with that partition
     */
    Map<Integer, PartitionMetadata> partitionsForTopic(String topic);

    /**
     * Get partition information of specific node.
     * The key is the TopicPartition, and the value is the metadata
     * associated with that partition.
     */
    Map<TopicPartition, PartitionMetadata> partitionsForNode(int nodeId);

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

    /**
     * Get ClusterResource that includes cluster id
     */
    ClusterResource clusterResource();

    /**
     * Given a partition metadata, return the subset of the replicas that are offline
     */
    List<Integer> 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 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 to BrokerRegistration which already includes the necessary information, making the ClusterMetadata implementation much more efficient.

public interface BrokerMetadata {

    /**
     * The broker node id
     */
    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.

The main difference is isAvailable(), as PartitionRegistration does not include this information. However, since Cluster#availablePartitionsForTopic allows users to query available partitions, providing a default method here for convenience.

public interface PartitionMetadata {

    /**
     * The node id of the node currently acting as a leader for this partition or -1 if there is no leader
     */
    int leader();

    /**
     * The complete set of replicas for this partition regardless of whether they are alive or up-to-date
     */
    List<Integer> replicas();

    /**
     * The subset of the replicas that are in sync, that is caught-up to the leader and ready to take over as leader if
     * the leader should fail. Note that there is no guarantee the first element is leader, please use leader()
     * to get the leader of partition.
     */
    List<Integer> inSyncReplicas();

    /**
     * Check if the partition is available
     */
    default boolean isAvailable() {
       return leader() != Node.noNode().id();
    }
}

Proposed Changes

In DynamicTopicClusterQuotaPublisher#onMetadataUpdate, invoke the new method introduced in this KIP. And since now we are using the new interface, all ClientQuotaCallback invocation in the current code base (except the tests) should be replaced with  ClientQuotaCallbackHandler.

@@ -52,8 +51,8 @@ class DynamicTopicClusterQuotaPublisher (
     try {
       quotaManagers.clientQuotaCallbackPlugin().ifPresent(plugin => {
         if (delta.topicsDelta() != null || delta.clusterDelta() != null) {
-          val cluster = MetadataCache.toCluster(clusterId, newImage)
-          if (plugin.get().updateClusterMetadata(cluster)) {
+          val clusterMetadata = new DefaultClusterMetadata(clusterId, newImage)
+          if (plugin.get().updateClusterMetadata(clusterMetadata)) {
             quotaManagers.fetch.updateQuotaMetricConfigs()
             quotaManagers.produce.updateQuotaMetricConfigs()
             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.

For example, the implementation of ClusterMetadata#partitionsForTopic(String) simply wraps the map returned by TopicImage.partitions().

metadataImage.topics().getTopic(topic).partitions().entrySet().stream()
        .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));

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

metadataImage.topics().getTopic(topicPartition.topic()).partitions().get(topicPartition.partition());

Other methods in ClusterMetadata will follow a similar pattern.

Compatibility, Deprecation, and Migration Plan

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

However, users are expected to implement the new interface ClientQuotaCallbackHandler. The old one - ClientQuotaCallback will be removed in a future major release.

Test Plan

We will rely on the existing CustomQuotaCallbackTest.java to ensure the deprecated 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 the new method updateClusterMetadata(ClusterMetadata) to the old interface ClientQuotaCallback, while doing this have a downside is that once we remove that deprecated function, it will break compilation for anyone who already implementing it.

  Code Block
  language java
  public interface ClientQuotaCallback extends Configurable { // ...skip ...// @Deprecated(since = "4.1", forRemoval = true) boolean updateClusterMetadata(Cluster cluster); default boolean updateClusterMetadata(ClusterMetadata clusterMetadata) { return updateClusterMetadata(toCluster(clusterMetadata)); } }