chia7712 commented on code in PR #23555:
URL: https://github.com/apache/kafka/pull/23555#discussion_r4429639117
##########
metadata/src/main/java/org/apache/kafka/metadata/publisher/DynamicClientQuotaPublisher.java:
##########
@@ -56,10 +96,61 @@ public void onMetadataUpdate(MetadataDelta delta,
MetadataImage newImage, Loader
ClientQuotasDelta clientQuotasDelta = delta.clientQuotasDelta();
if (clientQuotasDelta != null) {
try {
- clientQuotaUpdater.accept(clientQuotasDelta);
+ clientQuotasDelta.changes().forEach(this::update);
} catch (Throwable t) {
faultHandler.handleFault("Uncaught exception while publishing
dynamic client quota changes from MetadataDelta up to " +
newImage.highestOffsetAndEpoch().offset(), t);
}
}
}
+
+ private void update(ClientQuotaEntity entity, ClientQuotaDelta quotaDelta)
{
+ Map<String, String> entries = entity.entries();
+ if (entries.containsKey(IP)) {
+ handleIpQuota(entity, quotaDelta);
+ } else if (entries.containsKey(USER) ||
entries.containsKey(CLIENT_ID)) {
+ handleUserClientQuota(entity, quotaDelta);
+ } else {
+ log.warn("Ignoring unsupported quota entity {}.", entity);
+ }
+ }
+
+ private void handleIpQuota(ClientQuotaEntity entity, ClientQuotaDelta
quotaDelta) {
+ // An empty Optional represents the default IP entity.
+ Optional<InetAddress> inetAddress =
Optional.ofNullable(entity.entries().get(IP)).map(ip -> {
Review Comment:
Should we evaluate the address lazily?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
