Status

Current state: "Under DiscussionCanceled"

Discussion thread: https://lists.apache.org/thread/9l6m1scv765ohbwjkwq5sotjcdjzjp4q

Vote thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA: KAFKA-18239

: https://lists.apache.org/thread/mlzvnnfx9jwv0v0lql1hocno4xxk7cr8

JIRA:

Please keep the discussion on 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.

boolean updateClusterMetadata(ClusterMetadata clusterMetadata);

Here is the whole content of the new interface:

public interface ClientQuotaCallbackHandler 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.* Quota callback invoked to determine the quota metric tags to be applied for a request.
     * Quota limits are associated with quota metrics and all clients which use the same
     * metric tags share the quota limit.
     *
     * This@param isquotaType usefulType ifof quota computationrequested
 takes partitions into account.
     * Topics@param thatprincipal areThe beinguser deletedprincipal willof notthe beconnection includedfor inwhich `cluster`.
quota is    *requested
     * @param clusterMetadataclientId Cluster metadataThe includingclient partitionsid andassociated theirwith leaders, if knownthe request
     * @return truequota ifmetric quotastags havethat changedindicate andwhich metricother configsclients mayshare need to be updatedthis quota
     */
    defaultMap<String, booleanString> updateClusterMetadata(ClusterMetadata clusterMetadata) {
   quotaMetricTags(ClientQuotaType quotaType, KafkaPrincipal principal, String clientId);

    /**
     throw* new UnsupportedOperationException();
    }
}

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 {

    /**Returns the quota limit associated with the provided metric tags. These tags were returned from
     * a previous call to {@link #quotaMetricTags(ClientQuotaType, KafkaPrincipal, String)}. This method is
     * This callback is invoked wheneverby therequota aremanagers changesto inobtain the clustercurrent metadata,quota suchlimit asapplied 
to a metric when the *first brokersrequest
 being added or removed, topics* beingusing createdthese ortags deleted, or partition leadership updates.
     * This is useful if quota computation takes partitions into accountis processed. It is also invoked after a quota update or cluster metadata change.
     * TopicsIf thatthe tags are beingno deletedlonger willin notuse beafter includedthe in `cluster`.
     *update, (e.g. this is a {user, client-id} quota metric
     * @deprecatedand pleasethe usequota {@link ClientQuotaCallback#updateClusterMetadata(ClusterMetadata)} instead
     * @param cluster Cluster metadata including partitions and their leaders, if knownnow in use is a {user} quota), null is returned.
     *
     * @return@param truequotaType if quotasType haveof changedquota andrequested
 metric configs may need to* be@param updated
metricTags Metric tags for a */
quota metric of type @Deprecated`quotaType`
    boolean updateClusterMetadata(Cluster cluster);
}

The ClusterMetadata used in ClientQuotaCallbackHandler#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 { * @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 mapupdate ofcallback brokers.
that is invoked when quota configuration *for Thean keyentity is
   the broker node* ID,updated andin the quorum. valueThis is the metadata useful to track configured quotas if built-in quota configuration
     * associated with that brokertools are used for quota management.
     */
    Map<Integer, BrokerMetadata> brokers();

    /**
 @param quotaType   *Type Getof specificquota brokerbeing information.updated
     */
 @param quotaEntity The Optional<BrokerMetadata>quota broker(int nodeId);

    /**entity for which quota is being updated
     * Get@param partitionnewValue information for a specific topic.The new quota value
     */
   The keyvoid is the partition ID, and the value is the metadataupdateQuota(ClientQuotaType quotaType, ClientQuotaEntity quotaEntity, double newValue);

    /**
     * Quota configuration associatedremoval withcallback that partition.
is invoked when quota configuration */
for an   Map<Integer, PartitionMetadata> partitionsForTopic(String topic);

    /**
     * Get partition information for a specific node.entity is
     * removed in the quorum. This is useful to track configured quotas if built-in quota configuration
     * Thetools keyare isused thefor TopicPartition, and the value is the metadataquota management.
     *
     * associated@param withquotaType that partition.
 Type of quota being */updated
    Map<TopicPartition, PartitionMetadata>* partitionsForNode(int nodeId);

    /**
     * Get specific partition information.@param quotaEntity The quota entity for which quota is being updated
     */
    Optional<PartitionMetadata>void partitionremoveQuota(TopicPartition topicPartitionClientQuotaType quotaType, ClientQuotaEntity quotaEntity);

    /**
     * Returns Gettrue aif mapany of topics.
the existing quota configs may *have Thebeen keyupdated issince the topiclast ID,call
 and the value is the* topicto name.
this method for the provided */
quota type. Quota updates Map<Uuid,as String> topics();

    /**a result of calls to
     * Get ClusterResource{@link #updateClusterMetadata(ClusterMetadata)}, which includes the cluster ID.{@link #updateQuota(ClientQuotaType, ClientQuotaEntity, double)}
     */
 and {@link  ClusterResource clusterResource();

    /**#removeQuota(ClientQuotaType, ClientQuotaEntity)} are automatically processed.
     * So Givencallbacks partitionthat metadata,rely returnonly theon subsetbuilt-in ofquota theconfiguration replicastools thatalways arereturn offlinefalse. Quota callbacks
     * Ifwith thereexternal isquota noconfiguration offlineor replicacustom inreconfigurable thequota partition,configs returnthat anaffect empty array.quota limits must
     */
    int[] offlineReplicas(PartitionMetadata partitionMetadata);
} 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.
     *
     * @param quotaType Type of quota
     */
    boolean quotaResetRequired(ClientQuotaType quotaType);

    /**
     * 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.
     *
     * @param clusterMetadata 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(ClusterMetadata clusterMetadata);

    /**
     * Closes this instance.
     */
    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, please use {@link 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 ClientQuotaCallbackHandler#updateClusterMetadata(ClusterMetadata)} instead
     * @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
     */
    @Deprecated
    boolean updateClusterMetadata(Cluster cluster);

    default boolean updateClusterMetadata(ClusterMetadata clusterMetadata) {
        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 the cluster.
     */
    List<Integer> brokerIds();

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

    /**
     * Get specific partition information.
     */
    Optional<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. 
     * Return an empty map if topic does not exist.
     */
    Map<Integer, PartitionMetadata> partitions(String topic);

    /**
     * Get partition information of specific topic and specific node.
     * The key is the partition id, and the value is the metadata
     * associated with that partition.
     * Return an empty map if the topic or node id does not exist.
     */
    Map<Integer, PartitionMetadata> partitions(String topic, 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.
     * Return an empty list if there is no offline replicas.
     */
    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 convenienceUnlike 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. Additionally, DynamicTopicClusterQuotaPublisher can access BrokerRegistration, and we can add a new interface that already includes the necessary information in BrokerRegistration, making the ClusterMetadata implementation much more efficient.

public interface BrokerMetadataPartitionMetadata {

    /**
     * The broker node ID
id of the node currently */
acting as a leader int id();

    /**
     * The listeners of the broker nodefor this partition or -1 if there is no leader
     */
    Map<String, Endpoint> listenersint leader();
}

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

    /**
     * The node IDsubset of the replicas that nodeare currentlyin actingsync, asthat theis leadercaught-up forto thisthe partitionleader orand -1ready ifto theretake isover noas leader. if
     */
 the leader should int leader();

    /**
     * The complete set of replicas for this partition, regardless of whether they are alive or up-to-datefail. Note that there is no guarantee the first element is leader, please use leader()
     * to get the leader of partition.
     */
    int[]List<Integer> replicasinSyncReplicas();

    /**
     * TheCheck subsetif ofthe replicaspartition thatis areavailable
 in sync, meaning they are up-to-date and ready to take over as the leader if*/
    default boolean isAvailable() {
     * the currentreturn leader fails.() != Node.noNode().id();
     */
    int[] inSyncReplicas();

    /**
     * Check if the partition is available.
     */
    default boolean isAvailable() {
 }
}

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 => {
       return leader() !=if Node.noNode()delta.idtopicsDelta();
    }
}

Proposed Changes

In DynamicTopicClusterQuotaPublisher#onMetadataUpdate,  determine which interface the user has implemented and decide which updateClusterMetadata method to call.

clientQuotaCallback match {
  case _: ClientQuotaCallback =>
!= 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.

For example, the implementation of

...

ClusterMetadata#partitions(String) simply wraps the map returned by TopicImage.partitions().

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.

metadataImage topicPartition) {
    return image.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 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  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

...

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

...

• , 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 boolean updateClusterMetadata(Cluster cluster); default boolean updateClusterMetadata(ClusterMetadata clusterMetadata) {

...

• return updateClusterMetadata(toCluster(clusterMetadata)

...

• ); } }

...