chia7712 commented on code in PR #18196:
URL: https://github.com/apache/kafka/pull/18196#discussion_r1909124360
##########
core/src/test/scala/integration/kafka/api/CustomQuotaCallbackTest.scala:
##########
@@ -38,11 +40,9 @@ import java.util.Properties
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
import java.{lang, util}
-import scala.collection.Seq
import scala.collection.mutable.ArrayBuffer
import scala.jdk.CollectionConverters._
-@Disabled("KAFKA-18213")
Review Comment:
nice to see this test gets fixed. Could you please close KAFKA-18213 as
duplicate?
##########
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])
Review Comment:
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`
##########
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<>();
Review Comment:
we can leverage existent method. for example:
```
return listeners.keySet().stream().flatMap(l -> node(l).stream()).toList();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
