chia7712 commented on code in PR #23555:
URL: https://github.com/apache/kafka/pull/23555#discussion_r4433702170
##########
metadata/src/main/java/org/apache/kafka/metadata/publisher/DynamicClientQuotaPublisher.java:
##########
@@ -56,10 +96,62 @@ 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) {
+ quotaDelta.changes().forEach((key, value) -> {
+ // The connection quota only understands the connection rate limit.
+ if (!key.equals(QuotaConfig.IP_CONNECTION_RATE_OVERRIDE_CONFIG)) {
Review Comment:
If only `IP_CONNECTION_RATE_OVERRIDE_CONFIG` will be handled here, could we
use `get` instead of iterating?
```java
private void handleIpQuota(ClientQuotaEntity entity, ClientQuotaDelta
quotaDelta) {
OptionalDouble value =
quotaDelta.changes().get(QuotaConfig.IP_CONNECTION_RATE_OVERRIDE_CONFIG);
quotaDelta.changes().keySet().stream()
.filter(key ->
!key.equals(QuotaConfig.IP_CONNECTION_RATE_OVERRIDE_CONFIG))
.forEach(key -> log.warn("Ignoring unexpected quota key {} for
entity {}", key, entity));
if (value == null) return;
Optional<InetAddress> inetAddress = ...resolve once...;
try {
ipQuotaUpdater.updateQuota(inetAddress, value.isPresent() ?
OptionalInt.of((int) value.getAsDouble()) : OptionalInt.empty());
} catch (Throwable t) {
log.error("Failed to update IP quota {}", entity, t);
}
}
```
--
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]
