m1a2st commented on code in PR #19068:
URL: https://github.com/apache/kafka/pull/19068#discussion_r2032907056
##########
core/src/main/java/kafka/server/QuotaFactory.java:
##########
@@ -113,22 +116,43 @@ public void shutdown() {
}
}
- public static QuotaManagers instantiate(KafkaConfig cfg, Metrics metrics,
Time time, String threadNamePrefix) {
- ClientQuotaCallback clientQuotaCallback = cfg.getConfiguredInstance(
- QuotaConfig.CLIENT_QUOTA_CALLBACK_CLASS_CONFIG,
ClientQuotaCallback.class);
+ public static QuotaManagers instantiate(
+ KafkaConfig cfg,
+ Metrics metrics,
+ Time time,
+ String threadNamePrefix,
+ String role
+ ) {
+ Optional<Plugin<ClientQuotaCallback>> clientQuotaCallbackPlugin =
createClientQuotaCallback(cfg, metrics, role);
+ Option<Plugin<ClientQuotaCallback>> clientQuotaCallbackPluginOption =
OptionConverters.toScala(clientQuotaCallbackPlugin);
return new QuotaManagers(
- new ClientQuotaManager(clientConfig(cfg), metrics,
QuotaType.FETCH, time, threadNamePrefix, Option.apply(clientQuotaCallback)),
- new ClientQuotaManager(clientConfig(cfg), metrics,
QuotaType.PRODUCE, time, threadNamePrefix, Option.apply(clientQuotaCallback)),
- new ClientRequestQuotaManager(clientConfig(cfg), metrics, time,
threadNamePrefix, Optional.ofNullable(clientQuotaCallback)),
- new
ControllerMutationQuotaManager(clientControllerMutationConfig(cfg), metrics,
time, threadNamePrefix, Option.apply(clientQuotaCallback)),
+ new ClientQuotaManager(clientConfig(cfg), metrics,
QuotaType.FETCH, time, threadNamePrefix, clientQuotaCallbackPluginOption),
+ new ClientQuotaManager(clientConfig(cfg), metrics,
QuotaType.PRODUCE, time, threadNamePrefix, clientQuotaCallbackPluginOption),
+ new ClientRequestQuotaManager(clientConfig(cfg), metrics, time,
threadNamePrefix, clientQuotaCallbackPlugin),
+ new
ControllerMutationQuotaManager(clientControllerMutationConfig(cfg), metrics,
time, threadNamePrefix, clientQuotaCallbackPluginOption),
new ReplicationQuotaManager(replicationConfig(cfg), metrics,
QuotaType.LEADER_REPLICATION, time),
new ReplicationQuotaManager(replicationConfig(cfg), metrics,
QuotaType.FOLLOWER_REPLICATION, time),
new ReplicationQuotaManager(alterLogDirsReplicationConfig(cfg),
metrics, QuotaType.ALTER_LOG_DIRS_REPLICATION, time),
- Optional.ofNullable(clientQuotaCallback)
+ clientQuotaCallbackPlugin.map(Plugin::get)
);
}
+ private static Optional<Plugin<ClientQuotaCallback>>
createClientQuotaCallback(
Review Comment:
We use `Optional<Plugin<ClientQuotaCallback>>` in the
`ClientRequestQuotaManager` constructor, so regardless of whether`
createClientQuotaCallback` returns an` Option` or an `Optional`, we still need
to convert it.
--
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
