Current state: "Under Discussion"
Discussion thread: [Change the link from the KIP proposal email archive to your own email thread]
JIRA: KAFKA-18239
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
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.
This KIP introduces a new method updateClusterMetadata(ClusterMetadata) 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 updateClusterMetadata(ClusterMetadata) have a default implementation and trigger the old one.
@@ -101,13 +114,82 @@ public interface ClientQuotaCallback extends Configurable {
* 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 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);
+ /**
+ * 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`.
+ *
+ * @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
+ */
+ default 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.
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 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 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 the 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.
*/
int[] replicas();
/**
* The subset of replicas that are in sync, meaning they are up-to-date and ready to take over as the leader if
* the current leader fails.
*/
int[] inSyncReplicas();
/**
* Check if the partition is available.
*/
default boolean isAvailable() {
return leader() != Node.noNode().id();
}
} |
In DynamicTopicClusterQuotaPublisher#onMetadataUpdate, invoke the new method introduced in this KIP.
@@ -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.
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.
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.
N/A