KAFKA-18225: ClientQuotaCallback#updateClusterMetadata is unsupported by kraft #18196

m1a2st · 2024-12-15T11:35:02Z

Jira: https://issues.apache.org/jira/browse/KAFKA-18225

we don't implement the ClientQuotaCallback#updateClusterMetadata in Kraft mode. We will implement it in 4.0 version to pass CustomQuotaCallbackTest test in Kraft mode.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

core/src/main/scala/kafka/server/metadata/DynamicClientQuotaPublisher.scala

TaiJuWu · 2024-12-15T12:57:23Z

core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala

@@ -565,3 +531,97 @@ class KRaftMetadataCache(
  }
 }

+object KRaftMetadataCache {
+
+  def toCluster(clusterId: String, image: MetadataImage): Cluster = {


This result is different with KraftMetadataCache#toCluster (the old will filter partitions base on listener).
If we decide to change this one, maybe we need to document it.

chia7712 · 2025-01-08T09:20:22Z

@m1a2st please fix the conflicts

# Conflicts: # core/src/test/scala/integration/kafka/api/CustomQuotaCallbackTest.scala

m1a2st · 2025-01-08T11:43:54Z

Thanks for @chia7712 reminder, resolve the conflict

chia7712

@m1a2st thanks for this fix!

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

chia7712 · 2025-01-09T16:31:53Z

core/src/main/scala/kafka/server/metadata/DynamicClientQuotaPublisher.scala

@@ -48,6 +51,11 @@ class DynamicClientQuotaPublisher(
  ): Unit = {
    val deltaName = s"MetadataDelta up to ${newImage.highestOffsetAndEpoch().offset}"
    try {
+        val clientQuotaCallback = conf.getConfiguredInstance(QuotaConfig.CLIENT_QUOTA_CALLBACK_CLASS_CONFIG, classOf[ClientQuotaCallback])


we should use the callback in quotaManagers rather than creating a new one! They are different instances and so this approach can update the callback used by quotaManagers

chia7712 · 2025-01-09T16:34:44Z

metadata/src/main/java/org/apache/kafka/metadata/BrokerRegistration.java

@@ -235,6 +235,13 @@ public Optional<Node> node(String listenerName) {
        }
        return Optional.of(new Node(id, endpoint.host(), endpoint.port(), rack.orElse(null), fenced));
    }
+
+    public List<Node> nodes() {
+        List<Node> nodes = new ArrayList<>();


we can leverage existent method. for example:

return listeners.keySet().stream().flatMap(l -> node(l).stream()).toList();

chia7712

@m1a2st thanks for this patch. I test your patch on my local and please fix following code also:

deleteTopic needs to use authorized admin
You must call removeQuotaOverrides after creating group1_user2. Otherwise, the removeQuota method of the custom callback will not be invoked. This is an interesting discrepancy. In ZooKeeper mode, removeQuota is called when altering SASL/SCRAM credentials. However, in Kraft mode, this behavior is absent. I'm uncertain whether this constitutes a breaking change. It appears to be an unusual behavior, as it is triggered by the addition of users. Instead of implementing this peculiar behavior, I suggest updating the documentation of the callback to reflect the actual implementation.
cc @dajac and @cmccabe

chia7712 · 2025-01-13T10:24:16Z

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

@@ -179,13 +212,12 @@ class CustomQuotaCallbackTest extends IntegrationTestHarness with SaslSetup {
  }

  private def createTopic(topic: String, numPartitions: Int, leader: Int): Unit = {
-    // TODO createTopic 
+    TestUtils.createTopicWithAdmin(createAdminClient(), topic, brokers, controllerServers, numPartitions)


You have to honor the leader to make replica leader hosted by correct broker - otherwise, this test will get flaky in the future

core/src/main/scala/kafka/server/metadata/DynamicClientQuotaPublisher.scala

chia7712

@m1a2st thanks for this patch. Could you please update the docs (https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/server/quota/ClientQuotaCallback.java#L26). It is not used by broker only

chia7712 · 2025-02-03T02:44:30Z

core/src/test/java/kafka/test/api/CustomQuotaCallbackTest.java

+
+    public static class CustomQuotaCallback implements ClientQuotaCallback {
+
+        public static AtomicInteger dynamicTopicCounter = new AtomicInteger(0);


Could you please distinguish the count for different node id? for example:

public static final Map<String, AtomicInteger> COUNTERS = new ConcurrentHashMap<>(); private String nodeId; @Override public void configure(Map<String, ?> configs) { nodeId = (String) configs.get("node.id"); } @Override public boolean updateClusterMetadata(Cluster cluster) { COUNTERS.computeIfAbsent(nodeId, k -> new AtomicInteger()).incrementAndGet(); return true; }

with this counter, we can ensure all controllers do call the callback.

chia7712

@m1a2st thanks for this patch. I leave some comments to fix the tests. PTAL

chia7712 · 2025-02-04T14:57:45Z

core/src/test/java/kafka/test/api/CustomQuotaCallbackTest.java

+            );
+
+            CustomQuotaCallback.COUNTERS.clear();
+            admin.alterClientQuotas(clientQuotaAlterations, new AlterClientQuotasOptions().validateOnly(true));


you have to make some changes to trigger the metadata update, so please remove validateOnly(true)

core/src/test/java/kafka/test/api/CustomQuotaCallbackTest.java

chia7712 · 2025-02-04T18:40:04Z

core/src/test/java/kafka/test/api/CustomQuotaCallbackTest.java

+            );
+            // Reset the counters, and we expect the callback to be triggered again in all controllers
+            CustomQuotaCallback.COUNTERS.clear();
+            admin.alterClientQuotas(clientQuotaAlterations, new AlterClientQuotasOptions());


I don't think this method can trigger the callback as it don't cause either topic or cluster change.

core/src/test/java/kafka/test/api/CustomQuotaCallbackTest.java

# Conflicts: # core/src/main/scala/kafka/server/MetadataCache.scala

mimaison

Thanks for the PR. I left a bunch of comments. It looks like the new test testCustomQuotaCallbackWithControllerServer() is consistently failing. Also this needs to be rebased to resolve the conflict.

mimaison · 2025-02-05T13:48:45Z

core/src/main/scala/kafka/server/metadata/DynamicTopicClusterQuotaPublisher.scala

+    delta: MetadataDelta,
+    newImage: MetadataImage,
+  ): Unit = {
+    val deltaName = s"MetadataDelta up to ${newImage.highestOffsetAndEpoch().offset}"


Can we only compute this val in the catch block?

mimaison · 2025-02-05T13:49:00Z

core/src/main/scala/kafka/server/metadata/DynamicTopicClusterQuotaPublisher.scala

+        s"publishing dynamic topic or cluster changes from $deltaName", t)
+    }
+  }
+}


Nit: missing new line

mimaison · 2025-02-05T13:51:37Z

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

     * Topics that are being deleted will not be included in `cluster`.
     *
-     * @param cluster Cluster metadata including partitions and their leaders if known
+     * @param cluster Cluster metadata including topic and cluster


I'm not sure I understand this change. I prefer the previous message.

mimaison · 2025-02-05T13:55:29Z

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

@@ -89,11 +89,11 @@ public interface ClientQuotaCallback extends Configurable {
    boolean quotaResetRequired(ClientQuotaType quotaType);

    /**
-     * Metadata update callback that is invoked whenever UpdateMetadata request is received from
-     * the controller. This is useful if quota computation takes partitions into account.
+     * Metadata update callback that is invoked whenever the topic and cluster delta changed. 


I'm not sure we want to mention cluster/topic deltas here, this is a public API and part of our javadoc. I think we can say something like: This callback is invoked whenever the cluster metadata changes. This includes brokers added or removed, topics created or deleted, and partition leadership changes.

mimaison · 2025-02-05T13:57:13Z

core/src/main/scala/kafka/server/MetadataCache.scala

@@ -144,4 +149,95 @@ object MetadataCache {
  ): KRaftMetadataCache = {
    new KRaftMetadataCache(brokerId, kraftVersionSupplier)
  }
+
+  def toCluster(clusterId: String, image: MetadataImage): Cluster = {
+    val brokerToNodes = new util.HashMap[Integer, java.util.List[Node]]


We can drop the java. prefix in java.util.List

mimaison · 2025-02-05T13:58:44Z

core/src/main/scala/kafka/server/MetadataCache.scala

+    val brokerToNodes = new util.HashMap[Integer, java.util.List[Node]]
+    image.cluster().brokers()
+      .values().stream()
+      .filter(broker => !broker.fenced())


-> .filter(!_.fenced()) but maybe you wrote it this way with the future conversion to Java in mind. In that case we can keep the code.

Yes, I want to keep this because we need to convert it to a Java future.

mimaison · 2025-02-05T13:58:57Z

core/src/main/scala/kafka/server/MetadataCache.scala

+      .filter(broker => !broker.fenced())
+      .forEach { broker => brokerToNodes.put(broker.id(), broker.nodes()) }
+
+    def getNodes(id: Int): java.util.List[Node] = brokerToNodes.get(id)


Again drop the java. prefix

mimaison · 2025-02-05T14:24:28Z

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

@@ -87,11 +90,25 @@ class CustomQuotaCallbackTest extends IntegrationTestHarness with SaslSetup {

  override def configureSecurityBeforeServersStart(testInfo: TestInfo): Unit = {
    super.configureSecurityBeforeServersStart(testInfo)
+  }


Do we still need to keep configureSecurityBeforeServersStart() if all it does is call the super implementation?

mimaison · 2025-02-05T14:31:45Z

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

  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
-  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly_ZK_implicit"))
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly"))


From my limited testing it looks like this test works with the new consumer too. Is there a reason you used getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly instead of getTestQuorumAndGroupProtocolParametersAll?

The main reason is that this test targets the server, not the consumer. Therefore, I didn't test different consumers separately, but I will address this.

mimaison · 2025-02-05T14:33:30Z

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

    createScramCredentials(createAdminClient(), JaasTestUtils.KAFKA_SCRAM_ADMIN, JaasTestUtils.KAFKA_SCRAM_ADMIN_PASSWORD)
  }

+  override def addFormatterSettings(formatter: Formatter): Unit = {
+    formatter.setScramArguments(
+      List(s"SCRAM-SHA-256=[name=${JaasTestUtils.KAFKA_SCRAM_ADMIN},password=${JaasTestUtils.KAFKA_SCRAM_ADMIN_PASSWORD}]").asJava)


We could use util.List.of() instead of asJava here

chia7712 · 2025-02-05T16:41:36Z

core/src/test/java/kafka/test/api/CustomQuotaCallbackTest.java

+
+        @Override
+        public Double quotaLimit(ClientQuotaType quotaType, Map<String, String> metricTags) {
+            return 0.0;


You have to return Double.MAX_VALUE ... otherwise, the ops can't get pass as they are rejected due to throttle.

chia7712 · 2025-02-05T16:44:19Z

It looks like the new test testCustomQuotaCallbackWithControllerServer() is consistently failing.

@mimaison that is caused by #18196 (comment)

m1a2st · 2025-02-06T12:14:42Z

Thanks for @chia7712 and @mimaison review, I address comments, PTAL

chia7712

LGTM - will merge it tomorrow

In combined mode, the callback will be triggered twice - not sure whether it is a potential issue

dajac · 2025-02-10T09:26:16Z

@chia7712 Thanks for the review. Can we merge it?

…by kraft (#18196) This commit ensures that the ClientQuotaCallback#updateClusterMetadata method is executed in KRaft mode. This method is triggered whenever a topic or cluster metadata change occurs. However, in KRaft mode, the current implementation of the updateClusterMetadata API is inefficient due to the requirement of creating a full Cluster object. To address this, a follow-up issue (KAFKA-18239) has been created to explore more efficient mechanisms for providing cluster information to the ClientQuotaCallback without incurring the overhead of a full Cluster object creation. Reviewers: Mickael Maison <[email protected]>, TaiJuWu <[email protected]>, Chia-Ping Tsai <[email protected]>

…ssume-baseline-3.3-for-server * apache-github/trunk: KAFKA-18366 Remove KafkaConfig.interBrokerProtocolVersion (apache#18820) KAFKA-18658 add import control for examples module (apache#18812) MINOR: Java version and TLS documentation improvements (apache#18822) KAFKA-18743 Remove leader.imbalance.per.broker.percentage as it is not supported by Kraft (apache#18821) KAFKA-18225 ClientQuotaCallback#updateClusterMetadata is unsupported by kraft (apache#18196) KAFKA-18441: Remove flaky tag on KafkaAdminClientTest#testAdminClientApisAuthenticationFailure (apache#18847) MINOR: fix KStream#to incorrect javadoc (apache#18838) KAFKA-18763: changed the assertion statement for acknowledgements to include only successful acks (apache#18846) MINOR: Accept specifying consumer group assignors by their short names (apache#18832)

…by kraft (apache#18196) This commit ensures that the ClientQuotaCallback#updateClusterMetadata method is executed in KRaft mode. This method is triggered whenever a topic or cluster metadata change occurs. However, in KRaft mode, the current implementation of the updateClusterMetadata API is inefficient due to the requirement of creating a full Cluster object. To address this, a follow-up issue (KAFKA-18239) has been created to explore more efficient mechanisms for providing cluster information to the ClientQuotaCallback without incurring the overhead of a full Cluster object creation. Reviewers: Mickael Maison <[email protected]>, TaiJuWu <[email protected]>, Chia-Ping Tsai <[email protected]>

m1a2st added 3 commits December 15, 2024 15:47

wip

5fe6d1a

add clusterId to DynamicClientQuotaPublisher

b3edea5

Draft a version

15d7dbb

github-actions bot added the triage PRs from the community label Dec 15, 2024

m1a2st changed the title ~~KAFKA-18225: ClientQuotaCallback#updateClusterMetadata is unsupported by kraft~~ [WIP]KAFKA-18225: ClientQuotaCallback#updateClusterMetadata is unsupported by kraft Dec 15, 2024

github-actions bot added core Kafka Broker kraft labels Dec 15, 2024

TaiJuWu reviewed Dec 15, 2024

View reviewed changes

core/src/main/scala/kafka/server/metadata/DynamicClientQuotaPublisher.scala Outdated Show resolved Hide resolved

TaiJuWu reviewed Dec 15, 2024

View reviewed changes

mumrah added the ci-approved label Dec 15, 2024

m1a2st added 4 commits December 17, 2024 00:12

revert test

4123c70

Merge branch 'trunk' into KAFKA-18225

765d7f1

add new test

513b381

update the test

74b1006

m1a2st changed the title ~~[WIP]KAFKA-18225: ClientQuotaCallback#updateClusterMetadata is unsupported by kraft~~ KAFKA-18225: ClientQuotaCallback#updateClusterMetadata is unsupported by kraft Dec 17, 2024

Merge branch 'trunk' into KAFKA-18225

d1cfe97

github-actions bot removed the triage PRs from the community label Dec 19, 2024

Merge branch 'trunk' into KAFKA-18225

a4758e7

m1a2st added 2 commits January 8, 2025 19:39

Merge branch 'trunk' into KAFKA-18225

dee9a65

# Conflicts: # core/src/test/scala/integration/kafka/api/CustomQuotaCallbackTest.scala

resolve conflict

cd853b6

chia7712 reviewed Jan 9, 2025

View reviewed changes

m1a2st added 5 commits January 10, 2025 01:08

Merge branch 'trunk' into KAFKA-18225

48a8b41

addressed by comments

6bf41a7

addressed by comments

3fe1d7c

Merge branch 'trunk' into KAFKA-18225

ed50ea8

fix var name

f87db91

chia7712 reviewed Jan 13, 2025

View reviewed changes

update delete topic

57fad78

m1a2st added 5 commits January 26, 2025 17:02

fix the conflict

c5fb945

fix the conflict

c986339

Merge branch 'trunk' into KAFKA-18225

9721c48

Merge branch 'trunk' into KAFKA-18225

294cb93

Merge branch 'trunk' into KAFKA-18225

3877941

chia7712 reviewed Feb 3, 2025

View reviewed changes

m1a2st added 2 commits February 3, 2025 20:09

Merge branch 'trunk' into KAFKA-18225

2b527b4

addressed by comment

3dc5afb

chia7712 reviewed Feb 4, 2025

View reviewed changes

m1a2st added 2 commits February 4, 2025 23:34

Merge branch 'trunk' into KAFKA-18225

9469bff

addressed by comment

71123fb

chia7712 reviewed Feb 4, 2025

View reviewed changes

m1a2st added 2 commits February 5, 2025 22:32

Merge branch 'trunk' into KAFKA-18225

d902055

# Conflicts: # core/src/main/scala/kafka/server/MetadataCache.scala

resolve the conflict

3673d78

mimaison reviewed Feb 5, 2025

View reviewed changes

Merge branch 'trunk' into KAFKA-18225

21cf98e

chia7712 reviewed Feb 5, 2025

View reviewed changes

update the test

9451d69

address all comments

ee23d92

update new line

7924c03

chia7712 approved these changes Feb 6, 2025

View reviewed changes

chia7712 merged commit 70adf74 into apache:trunk Feb 10, 2025
9 checks passed

chia7712 mentioned this pull request Feb 17, 2025

KAFKA-18213: Convert CustomQuotaCallbackTest to KRaft #18126

Closed

3 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-18225: ClientQuotaCallback#updateClusterMetadata is unsupported by kraft #18196

KAFKA-18225: ClientQuotaCallback#updateClusterMetadata is unsupported by kraft #18196

m1a2st commented Dec 15, 2024

TaiJuWu Dec 15, 2024 •

edited

Loading

chia7712 commented Jan 8, 2025

m1a2st commented Jan 8, 2025

chia7712 left a comment

chia7712 Jan 9, 2025

chia7712 Jan 9, 2025

chia7712 left a comment

chia7712 Jan 13, 2025

chia7712 left a comment

chia7712 Feb 3, 2025

chia7712 left a comment

chia7712 Feb 4, 2025

chia7712 Feb 4, 2025

mimaison left a comment

mimaison Feb 5, 2025

mimaison Feb 5, 2025

mimaison Feb 5, 2025

mimaison Feb 5, 2025

mimaison Feb 5, 2025

mimaison Feb 5, 2025

m1a2st Feb 6, 2025

mimaison Feb 5, 2025

mimaison Feb 5, 2025

mimaison Feb 5, 2025

m1a2st Feb 6, 2025

mimaison Feb 5, 2025

chia7712 Feb 5, 2025

chia7712 commented Feb 5, 2025

m1a2st commented Feb 6, 2025

chia7712 left a comment

dajac commented Feb 10, 2025


		public static class CustomQuotaCallback implements ClientQuotaCallback {

		public static AtomicInteger dynamicTopicCounter = new AtomicInteger(0);

KAFKA-18225: ClientQuotaCallback#updateClusterMetadata is unsupported by kraft #18196

KAFKA-18225: ClientQuotaCallback#updateClusterMetadata is unsupported by kraft #18196

Conversation

m1a2st commented Dec 15, 2024

Committer Checklist (excluded from commit message)

TaiJuWu Dec 15, 2024 • edited Loading

Choose a reason for hiding this comment

chia7712 commented Jan 8, 2025

m1a2st commented Jan 8, 2025

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mimaison left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 commented Feb 5, 2025

m1a2st commented Feb 6, 2025

chia7712 left a comment

Choose a reason for hiding this comment

dajac commented Feb 10, 2025

TaiJuWu Dec 15, 2024 •

edited

Loading