You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mimaison (via GitHub)" <gi...@apache.org> on 2025/02/27 17:32:48 UTC
[PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
mimaison opened a new pull request, #19050:
URL: https://github.com/apache/kafka/pull/19050
This also adds metrics to StandardAuthorizer
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2041136887
##########
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java:
##########
@@ -40,14 +47,46 @@ private Plugin(T instance, PluginMetricsImpl pluginMetrics) {
this.pluginMetrics = Optional.ofNullable(pluginMetrics);
}
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param tagsSupplier supplier to retrieve the tags
+ * @return the plugin
+ */
+ public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, Supplier<Map<String, String>> tagsSupplier) {
+ PluginMetricsImpl pluginMetrics = null;
+ if (instance instanceof Monitorable && metrics != null) {
+ pluginMetrics = new PluginMetricsImpl(metrics, tagsSupplier.get());
+ ((Monitorable) instance).withPluginMetrics(pluginMetrics);
+ }
+ return new Plugin<>(instance, pluginMetrics);
+ }
+
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param key the value for the <code>config</code> tag
+ * @return the plugin
+ */
public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, String key) {
return wrapInstance(instance, metrics, () -> tags(key, instance));
}
- public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, String key, Map<String, String> extraTags) {
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param name extra tag name to add
+ * @param value extra tag value to add
+ * @param key the value for the <code>config</code> tag
+ * @return the plugin
+ */
+ public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, String key, String name, String value) {
Review Comment:
At the moment all the callers have a single extra tag, so it's simpler than having them build a `LinkedHashMap`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison merged PR #19050:
URL: https://github.com/apache/kafka/pull/19050
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2031382775
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3284,7 +3285,7 @@ private boolean maybeUpdateRegularExpressions(
* @param log The log instance.
* @param time The time instance.
* @param image The metadata image to use for listing the topics.
- * @param authorizer The authorizer.
+ * @param authorizerPlugin The authorizer.
Review Comment:
I think it's a bit overkill. This field was an `Optional<Authorizer>` and we just said `The authorizer`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2041137804
##########
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java:
##########
@@ -40,14 +47,46 @@ private Plugin(T instance, PluginMetricsImpl pluginMetrics) {
this.pluginMetrics = Optional.ofNullable(pluginMetrics);
}
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param tagsSupplier supplier to retrieve the tags
+ * @return the plugin
+ */
+ public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, Supplier<Map<String, String>> tagsSupplier) {
+ PluginMetricsImpl pluginMetrics = null;
+ if (instance instanceof Monitorable && metrics != null) {
+ pluginMetrics = new PluginMetricsImpl(metrics, tagsSupplier.get());
+ ((Monitorable) instance).withPluginMetrics(pluginMetrics);
+ }
+ return new Plugin<>(instance, pluginMetrics);
+ }
Review Comment:
Maybe we should update the `PluginMetrics` API to require a `LinkedHashMap`, or at least mention it in the javadoc.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2031012460
##########
metadata/src/main/java/org/apache/kafka/metadata/publisher/AclPublisher.java:
##########
@@ -103,6 +105,9 @@ public void onMetadataUpdate(MetadataDelta delta, MetadataImage newImage, Loader
@Override
public void close() {
- authorizer.ifPresent(clusterMetadataAuthorizer -> clusterMetadataAuthorizer.completeInitialLoad(new TimeoutException()));
+ authorizer.ifPresent(authorizer -> {
+ ClusterMetadataAuthorizer clusterMetadataAuthorizer = (ClusterMetadataAuthorizer) authorizer.get();
+ clusterMetadataAuthorizer.completeInitialLoad(new TimeoutException());
+ });
}
}
Review Comment:
I was already missing. Added
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2028375807
##########
metadata/src/main/java/org/apache/kafka/metadata/publisher/AclPublisher.java:
##########
@@ -103,6 +105,9 @@ public void onMetadataUpdate(MetadataDelta delta, MetadataImage newImage, Loader
@Override
public void close() {
- authorizer.ifPresent(clusterMetadataAuthorizer -> clusterMetadataAuthorizer.completeInitialLoad(new TimeoutException()));
+ authorizer.ifPresent(authorizer -> {
+ ClusterMetadataAuthorizer clusterMetadataAuthorizer = (ClusterMetadataAuthorizer) authorizer.get();
+ clusterMetadataAuthorizer.completeInitialLoad(new TimeoutException());
+ });
}
}
Review Comment:
nit: new line
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2028335989
##########
jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/KRaftMetadataRequestBenchmark.java:
##########
@@ -94,8 +94,8 @@
@State(Scope.Benchmark)
@Fork(value = 1)
-@Warmup(iterations = 5)
-@Measurement(iterations = 15)
+@Warmup(iterations = 1)
+@Measurement(iterations = 3)
Review Comment:
No this is unwanted, I just used these smaller numbers to run the benchmark more quickly. I'll revert 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2041847685
##########
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java:
##########
@@ -40,14 +47,46 @@ private Plugin(T instance, PluginMetricsImpl pluginMetrics) {
this.pluginMetrics = Optional.ofNullable(pluginMetrics);
}
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param tagsSupplier supplier to retrieve the tags
+ * @return the plugin
+ */
+ public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, Supplier<Map<String, String>> tagsSupplier) {
+ PluginMetricsImpl pluginMetrics = null;
+ if (instance instanceof Monitorable && metrics != null) {
+ pluginMetrics = new PluginMetricsImpl(metrics, tagsSupplier.get());
+ ((Monitorable) instance).withPluginMetrics(pluginMetrics);
+ }
+ return new Plugin<>(instance, pluginMetrics);
+ }
Review Comment:
Thanks!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2030844573
##########
server/src/test/java/org/apache/kafka/server/MonitorablePluginsIntegrationTest.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Measurable;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Monitorable;
+import org.apache.kafka.common.metrics.PluginMetrics;
+import org.apache.kafka.common.replica.RackAwareReplicaSelector;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.metadata.authorizer.StandardAuthorizer;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.apache.kafka.clients.admin.AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG;
+import static org.apache.kafka.server.config.ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG;
+import static org.apache.kafka.server.config.ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class MonitorablePluginsIntegrationTest {
+
+ private static int controllerId(Type type) {
+ return type == Type.KRAFT ? 3000 : 0;
+ }
+
+ private static Map<String, String> expectedTags(String config, String clazz) {
+ return expectedTags(config, clazz, Map.of());
+ }
+
+ private static Map<String, String> expectedTags(String config, String clazz, Map<String, String> extraTags) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("config", config);
+ tags.put("class", clazz);
+ tags.putAll(extraTags);
+ return tags;
+ }
+
+ @ClusterTest(
+ types = {Type.KRAFT, Type.CO_KRAFT},
+ serverProperties = {
+ @ClusterConfigProperty(key = StandardAuthorizer.SUPER_USERS_CONFIG, value = "User:ANONYMOUS"),
+ @ClusterConfigProperty(key = AUTHORIZER_CLASS_NAME_CONFIG, value = "org.apache.kafka.metadata.authorizer.StandardAuthorizer"),
+ @ClusterConfigProperty(key = REPLICA_SELECTOR_CLASS_CONFIG, value = "org.apache.kafka.server.MonitorablePluginsIntegrationTest$MonitorableReplicaSelector")
+ }
+ )
+ public void testMonitorableServerPlugins(ClusterInstance clusterInstance) throws Exception {
+ try (Admin admin = Admin.create(Map.of(BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers()))) {
+ admin.describeCluster().clusterId().get();
+ }
Review Comment:
Right we don't need it now, I had it because I was also testing other server plugins. I'll delete it for now.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2030840916
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java:
##########
@@ -505,8 +506,8 @@ public Builder withShareGroupAssignor(ShareGroupPartitionAssignor shareGroupAssi
return this;
}
- public Builder withAuthorizer(Authorizer authorizer) {
- this.authorizer = Optional.of(authorizer);
+ public Builder withAuthorizerPlugin(Plugin<Authorizer> authorizerPlugin) {
+ this.authorizerPlugin = Optional.of(authorizerPlugin);
Review Comment:
As far as I can tell the other fields are also always set, so I'll leave the `null` check for consistency.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2041841871
##########
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java:
##########
@@ -40,14 +47,46 @@ private Plugin(T instance, PluginMetricsImpl pluginMetrics) {
this.pluginMetrics = Optional.ofNullable(pluginMetrics);
}
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param tagsSupplier supplier to retrieve the tags
+ * @return the plugin
+ */
+ public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, Supplier<Map<String, String>> tagsSupplier) {
+ PluginMetricsImpl pluginMetrics = null;
+ if (instance instanceof Monitorable && metrics != null) {
+ pluginMetrics = new PluginMetricsImpl(metrics, tagsSupplier.get());
+ ((Monitorable) instance).withPluginMetrics(pluginMetrics);
+ }
+ return new Plugin<>(instance, pluginMetrics);
+ }
Review Comment:
Sure, Open https://issues.apache.org/jira/browse/KAFKA-19139 to trace 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2042391494
##########
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java:
##########
@@ -40,14 +47,46 @@ private Plugin(T instance, PluginMetricsImpl pluginMetrics) {
this.pluginMetrics = Optional.ofNullable(pluginMetrics);
}
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param tagsSupplier supplier to retrieve the tags
+ * @return the plugin
+ */
+ public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, Supplier<Map<String, String>> tagsSupplier) {
+ PluginMetricsImpl pluginMetrics = null;
+ if (instance instanceof Monitorable && metrics != null) {
+ pluginMetrics = new PluginMetricsImpl(metrics, tagsSupplier.get());
+ ((Monitorable) instance).withPluginMetrics(pluginMetrics);
+ }
+ return new Plugin<>(instance, pluginMetrics);
+ }
Review Comment:
Should we depend on the LinkedHashMap implementation?
We can do this to guarantee the order while keeping the original API, and mention in the docs.
```
Map<String, String> metricsTags = new LinkedHashMap<>(this.tags);
metricsTags.putAll(new LinkedHashMap<>(tags));
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2028015016
##########
server/src/test/java/org/apache/kafka/server/MonitorablePluginsIntegrationTest.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Measurable;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Monitorable;
+import org.apache.kafka.common.metrics.PluginMetrics;
+import org.apache.kafka.common.replica.RackAwareReplicaSelector;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.metadata.authorizer.StandardAuthorizer;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.apache.kafka.clients.admin.AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG;
+import static org.apache.kafka.server.config.ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG;
+import static org.apache.kafka.server.config.ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class MonitorablePluginsIntegrationTest {
+
+ private static int controllerId(Type type) {
+ return type == Type.KRAFT ? 3000 : 0;
+ }
+
+ private static Map<String, String> expectedTags(String config, String clazz) {
+ return expectedTags(config, clazz, Map.of());
+ }
+
+ private static Map<String, String> expectedTags(String config, String clazz, Map<String, String> extraTags) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("config", config);
+ tags.put("class", clazz);
+ tags.putAll(extraTags);
+ return tags;
+ }
+
+ @ClusterTest(
+ types = {Type.KRAFT, Type.CO_KRAFT},
+ serverProperties = {
+ @ClusterConfigProperty(key = StandardAuthorizer.SUPER_USERS_CONFIG, value = "User:ANONYMOUS"),
+ @ClusterConfigProperty(key = AUTHORIZER_CLASS_NAME_CONFIG, value = "org.apache.kafka.metadata.authorizer.StandardAuthorizer"),
+ @ClusterConfigProperty(key = REPLICA_SELECTOR_CLASS_CONFIG, value = "org.apache.kafka.server.MonitorablePluginsIntegrationTest$MonitorableReplicaSelector")
+ }
+ )
+ public void testMonitorableServerPlugins(ClusterInstance clusterInstance) throws Exception {
+ try (Admin admin = Admin.create(Map.of(BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers()))) {
+ admin.describeCluster().clusterId().get();
+ }
Review Comment:
I think we don't need to descibe the cluster information in this test, right?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2027327151
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -766,7 +766,7 @@ class BrokerServer(
CoreUtils.swallow(dataPlaneRequestHandlerPool.shutdown(), this)
if (dataPlaneRequestProcessor != null)
CoreUtils.swallow(dataPlaneRequestProcessor.close(), this)
- authorizer.foreach(Utils.closeQuietly(_, "authorizer"))
+ authorizerPlugin.foreach(Utils.closeQuietly(_, "authorizer"))
Review Comment:
Should we rename "authorizer" to "authorizer plugin" for clarity? Since we explicitly close the authorizer in `Plugin#close()`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -250,7 +251,7 @@ public GroupCoordinatorShard build() {
throw new IllegalArgumentException("TopicPartition must be set.");
if (groupConfigManager == null)
throw new IllegalArgumentException("GroupConfigManager must be set.");
- if (authorizer == null)
+ if (authorizerPlugin == null)
Review Comment:
Should we also check `authorizerPlugin.isEmpty()`?
##########
core/src/main/scala/kafka/server/ControllerServer.scala:
##########
@@ -467,7 +470,7 @@ class ControllerServer(
CoreUtils.swallow(quotaManagers.shutdown(), this)
Utils.closeQuietly(controller, "controller")
Utils.closeQuietly(quorumControllerMetrics, "quorum controller metrics")
- authorizer.foreach(Utils.closeQuietly(_, "authorizer"))
+ authorizerPlugin.foreach(Utils.closeQuietly(_, "authorizer"))
Review Comment:
ditto: Should we rename "authorizer" to "authorizer plugin" for clarity? Since we explicitly close the authorizer in `Plugin#close()`.
##########
jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/KRaftMetadataRequestBenchmark.java:
##########
@@ -94,8 +94,8 @@
@State(Scope.Benchmark)
@Fork(value = 1)
-@Warmup(iterations = 5)
-@Measurement(iterations = 15)
+@Warmup(iterations = 1)
+@Measurement(iterations = 3)
Review Comment:
Just curious, why is this change necessary?
##########
metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizer.java:
##########
@@ -47,7 +54,7 @@
* The standard authorizer which is used in KRaft-based clusters if no other authorizer is
* configured.
Review Comment:
We can remove KRaft-related keywords to improve this document. because we only have KRaft based cluster in 4.X.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java:
##########
@@ -505,8 +506,8 @@ public Builder withShareGroupAssignor(ShareGroupPartitionAssignor shareGroupAssi
return this;
}
- public Builder withAuthorizer(Authorizer authorizer) {
- this.authorizer = Optional.of(authorizer);
+ public Builder withAuthorizerPlugin(Plugin<Authorizer> authorizerPlugin) {
+ this.authorizerPlugin = Optional.of(authorizerPlugin);
Review Comment:
We set the parameter is `Optional<Plugin<Authorizer>> authorizerPlugin` in `GroupMetadataManager`, If we consistently pass `Plugin<Authorizer>` into the builder and wrap it inside the method, can we avoid the need to check for null?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2030836809
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -250,7 +251,7 @@ public GroupCoordinatorShard build() {
throw new IllegalArgumentException("TopicPartition must be set.");
if (groupConfigManager == null)
throw new IllegalArgumentException("GroupConfigManager must be set.");
- if (authorizer == null)
+ if (authorizerPlugin == null)
Review Comment:
It's fine to pass `Optional.empty()` down the call stack.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on PR #19050:
URL: https://github.com/apache/kafka/pull/19050#issuecomment-2797321585
> Overall LGTM. Will you be addressing [#19068 (comment)](https://github.com/apache/kafka/pull/19068#discussion_r2033623742) in this PR?
I pushed a commit to address 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2041132665
##########
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java:
##########
@@ -40,14 +47,46 @@ private Plugin(T instance, PluginMetricsImpl pluginMetrics) {
this.pluginMetrics = Optional.ofNullable(pluginMetrics);
}
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param tagsSupplier supplier to retrieve the tags
+ * @return the plugin
+ */
+ public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, Supplier<Map<String, String>> tagsSupplier) {
+ PluginMetricsImpl pluginMetrics = null;
+ if (instance instanceof Monitorable && metrics != null) {
+ pluginMetrics = new PluginMetricsImpl(metrics, tagsSupplier.get());
+ ((Monitorable) instance).withPluginMetrics(pluginMetrics);
+ }
+ return new Plugin<>(instance, pluginMetrics);
+ }
+
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param key the value for the <code>config</code> tag
+ * @return the plugin
+ */
public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, String key) {
return wrapInstance(instance, metrics, () -> tags(key, instance));
}
- public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, String key, Map<String, String> extraTags) {
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param name extra tag name to add
+ * @param value extra tag value to add
+ * @param key the value for the <code>config</code> tag
+ * @return the plugin
+ */
+ public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, String key, String name, String value) {
Review Comment:
Would it be better to use a `LinkedHashMap` instead of separate String name/value pairs?
##########
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java:
##########
@@ -40,14 +47,46 @@ private Plugin(T instance, PluginMetricsImpl pluginMetrics) {
this.pluginMetrics = Optional.ofNullable(pluginMetrics);
}
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
Review Comment:
ditto I think we should also mention that the instance's class name will be used as the value for the `<code>class</code>` tag.
##########
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java:
##########
@@ -40,14 +47,46 @@ private Plugin(T instance, PluginMetricsImpl pluginMetrics) {
this.pluginMetrics = Optional.ofNullable(pluginMetrics);
}
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param tagsSupplier supplier to retrieve the tags
+ * @return the plugin
+ */
+ public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, Supplier<Map<String, String>> tagsSupplier) {
+ PluginMetricsImpl pluginMetrics = null;
+ if (instance instanceof Monitorable && metrics != null) {
+ pluginMetrics = new PluginMetricsImpl(metrics, tagsSupplier.get());
+ ((Monitorable) instance).withPluginMetrics(pluginMetrics);
+ }
+ return new Plugin<>(instance, pluginMetrics);
+ }
+
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param key the value for the <code>config</code> tag
+ * @return the plugin
+ */
public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, String key) {
return wrapInstance(instance, metrics, () -> tags(key, instance));
}
- public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, String key, Map<String, String> extraTags) {
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
Review Comment:
I think we should also mention that the instance's class name will be used as the value for the <code>class</code> tag.
##########
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java:
##########
@@ -40,14 +47,46 @@ private Plugin(T instance, PluginMetricsImpl pluginMetrics) {
this.pluginMetrics = Optional.ofNullable(pluginMetrics);
}
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param tagsSupplier supplier to retrieve the tags
+ * @return the plugin
+ */
+ public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, Supplier<Map<String, String>> tagsSupplier) {
+ PluginMetricsImpl pluginMetrics = null;
+ if (instance instanceof Monitorable && metrics != null) {
+ pluginMetrics = new PluginMetricsImpl(metrics, tagsSupplier.get());
+ ((Monitorable) instance).withPluginMetrics(pluginMetrics);
+ }
+ return new Plugin<>(instance, pluginMetrics);
+ }
+
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
Review Comment:
ditto I think we should also mention that the instance's class name will be used as the value for the `<code>class</code>` tag.
##########
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java:
##########
@@ -40,14 +47,46 @@ private Plugin(T instance, PluginMetricsImpl pluginMetrics) {
this.pluginMetrics = Optional.ofNullable(pluginMetrics);
}
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param tagsSupplier supplier to retrieve the tags
+ * @return the plugin
+ */
+ public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, Supplier<Map<String, String>> tagsSupplier) {
+ PluginMetricsImpl pluginMetrics = null;
+ if (instance instanceof Monitorable && metrics != null) {
+ pluginMetrics = new PluginMetricsImpl(metrics, tagsSupplier.get());
+ ((Monitorable) instance).withPluginMetrics(pluginMetrics);
+ }
+ return new Plugin<>(instance, pluginMetrics);
+ }
Review Comment:
Also ask the method PluginMetricsImpl#metricName, it also use `Map.putAll()`, Should we also update it for order?
```
@Override
public MetricName metricName(String name, String description, Map<String, String> tags) {
if (closing) throw new IllegalStateException("This PluginMetrics instance is closed");
for (String tagName : tags.keySet()) {
if (this.tags.containsKey(tagName)) {
throw new IllegalArgumentException("Cannot use " + tagName + " as a tag name");
}
}
Map<String, String> metricsTags = new LinkedHashMap<>(this.tags);
metricsTags.putAll(tags);
return metrics.metricName(name, GROUP, description, metricsTags);
}
```
##########
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java:
##########
@@ -60,6 +99,13 @@ private static <T> Map<String, String> tags(String key, T instance) {
return tags;
}
+ /**
+ * Wrap a list of instances into Plugins.
+ * @param instances the instances to wrap
Review Comment:
ditto I think we should also mention that the instance's class name will be used as the value for the `<code>class</code>` tag.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2027138995
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -2654,7 +2655,7 @@ class ReplicaManager(val config: KafkaConfig,
config.replicaSelectorClassName.map { className =>
val tmpReplicaSelector: ReplicaSelector = Utils.newInstance(className, classOf[ReplicaSelector])
tmpReplicaSelector.configure(config.originals())
- Plugin.wrapInstance(tmpReplicaSelector, metrics, className)
+ Plugin.wrapInstance(tmpReplicaSelector, metrics, ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG)
Review Comment:
This fixes an issue we introduced in https://github.com/apache/kafka/commit/79fe1305b62498cc6e9a15b2163134f0df3e1757
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on PR #19050:
URL: https://github.com/apache/kafka/pull/19050#issuecomment-2775985730
@Yunyung Sorry the PR was not fully complete. I updated the javadoc and added an integration test. It should be ready to review now.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2031390658
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3284,7 +3285,7 @@ private boolean maybeUpdateRegularExpressions(
* @param log The log instance.
* @param time The time instance.
* @param image The metadata image to use for listing the topics.
- * @param authorizer The authorizer.
+ * @param authorizerPlugin The authorizer.
Review Comment:
Thanks, make sense to me.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2041132013
##########
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java:
##########
@@ -40,14 +47,46 @@ private Plugin(T instance, PluginMetricsImpl pluginMetrics) {
this.pluginMetrics = Optional.ofNullable(pluginMetrics);
}
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param tagsSupplier supplier to retrieve the tags
+ * @return the plugin
+ */
+ public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, Supplier<Map<String, String>> tagsSupplier) {
+ PluginMetricsImpl pluginMetrics = null;
+ if (instance instanceof Monitorable && metrics != null) {
+ pluginMetrics = new PluginMetricsImpl(metrics, tagsSupplier.get());
+ ((Monitorable) instance).withPluginMetrics(pluginMetrics);
+ }
+ return new Plugin<>(instance, pluginMetrics);
+ }
+
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param key the value for the <code>config</code> tag
+ * @return the plugin
+ */
public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, String key) {
return wrapInstance(instance, metrics, () -> tags(key, instance));
}
- public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, String key, Map<String, String> extraTags) {
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
Review Comment:
I think we should also mention that the instance's class name will be used as the value for the `<code>class</code>` tag.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2041137300
##########
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java:
##########
@@ -60,6 +99,13 @@ private static <T> Map<String, String> tags(String key, T instance) {
return tags;
}
+ /**
+ * Wrap a list of instances into Plugins.
+ * @param instances the instances to wrap
Review Comment:
This is already mentioned in the class javadoc
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2044022395
##########
metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizer.java:
##########
@@ -207,4 +216,45 @@ static AuthorizationResult getDefaultResult(Map<String, ?> configs) {
if (configValue == null) return DENIED;
return Boolean.parseBoolean(configValue.toString().trim()) ? ALLOWED : DENIED;
}
+
+ @Override
+ public void withPluginMetrics(PluginMetrics metrics) {
+ this.authorizerMetrics = new AuthorizerMetrics(metrics);
+ }
+
+ private class AuthorizerMetrics {
Review Comment:
Have you considered exposing `AuthorizerMetrics` as public APIs for custom Authorizer?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2044525835
##########
metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizer.java:
##########
@@ -207,4 +216,45 @@ static AuthorizationResult getDefaultResult(Map<String, ?> configs) {
if (configValue == null) return DENIED;
return Boolean.parseBoolean(configValue.toString().trim()) ? ALLOWED : DENIED;
}
+
+ @Override
+ public void withPluginMetrics(PluginMetrics metrics) {
+ this.authorizerMetrics = new AuthorizerMetrics(metrics);
+ }
+
+ private class AuthorizerMetrics {
+ private final Sensor authorizationAllowedSensor;
+ private final Sensor authorizationDeniedSensor;
+ private final Sensor authorizationRequestSensor;
+
+ private AuthorizerMetrics(PluginMetrics metrics) {
+ authorizationAllowedSensor = metrics.addSensor("authorizer-authorization-allowed");
Review Comment:
That was a typo in the KIP, updated.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on PR #19050:
URL: https://github.com/apache/kafka/pull/19050#issuecomment-2804287543
@chia7712 If you have time, can you take a look? Thanks
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2042391494
##########
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java:
##########
@@ -40,14 +47,46 @@ private Plugin(T instance, PluginMetricsImpl pluginMetrics) {
this.pluginMetrics = Optional.ofNullable(pluginMetrics);
}
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param tagsSupplier supplier to retrieve the tags
+ * @return the plugin
+ */
+ public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, Supplier<Map<String, String>> tagsSupplier) {
+ PluginMetricsImpl pluginMetrics = null;
+ if (instance instanceof Monitorable && metrics != null) {
+ pluginMetrics = new PluginMetricsImpl(metrics, tagsSupplier.get());
+ ((Monitorable) instance).withPluginMetrics(pluginMetrics);
+ }
+ return new Plugin<>(instance, pluginMetrics);
+ }
Review Comment:
Should we depend on the LinkedHashMap implementation?
We can do this to guarantee the order while keeping the original API, and mention in the docs.
```
Map<String, String> metricsTags = new LinkedHashMap<>(this.tags);
metricsTags.putAll(new LinkedHashMap<>(tags));
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2027209374
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -2654,7 +2655,7 @@ class ReplicaManager(val config: KafkaConfig,
config.replicaSelectorClassName.map { className =>
val tmpReplicaSelector: ReplicaSelector = Utils.newInstance(className, classOf[ReplicaSelector])
tmpReplicaSelector.configure(config.originals())
- Plugin.wrapInstance(tmpReplicaSelector, metrics, className)
+ Plugin.wrapInstance(tmpReplicaSelector, metrics, ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG)
Review Comment:
Thanks for fix 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2044524561
##########
server/src/test/java/org/apache/kafka/server/MonitorablePluginsIntegrationTest.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server;
+
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Measurable;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Monitorable;
+import org.apache.kafka.common.metrics.PluginMetrics;
+import org.apache.kafka.common.replica.RackAwareReplicaSelector;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.metadata.authorizer.StandardAuthorizer;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.apache.kafka.server.config.ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG;
+import static org.apache.kafka.server.config.ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class MonitorablePluginsIntegrationTest {
+
+ private static int controllerId(Type type) {
+ return type == Type.KRAFT ? 3000 : 0;
+ }
+
+ private static Map<String, String> expectedTags(String config, String clazz) {
+ return expectedTags(config, clazz, Map.of());
+ }
+
+ private static Map<String, String> expectedTags(String config, String clazz, Map<String, String> extraTags) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("config", config);
+ tags.put("class", clazz);
+ tags.putAll(extraTags);
+ return tags;
+ }
+
+ @ClusterTest(
+ types = {Type.KRAFT, Type.CO_KRAFT},
+ serverProperties = {
+ @ClusterConfigProperty(key = StandardAuthorizer.SUPER_USERS_CONFIG, value = "User:ANONYMOUS"),
+ @ClusterConfigProperty(key = AUTHORIZER_CLASS_NAME_CONFIG, value = "org.apache.kafka.metadata.authorizer.StandardAuthorizer"),
+ @ClusterConfigProperty(key = REPLICA_SELECTOR_CLASS_CONFIG, value = "org.apache.kafka.server.MonitorablePluginsIntegrationTest$MonitorableReplicaSelector")
+ }
+ )
+ public void testMonitorableServerPlugins(ClusterInstance clusterInstance) {
+ assertAuthorizerMetrics(clusterInstance);
+ assertReplicaSelectorMetrics(clusterInstance);
+ }
+
+ private void assertAuthorizerMetrics(ClusterInstance clusterInstance) {
+ assertMetrics(
+ clusterInstance.brokers().get(0).metrics(),
+ 4,
+ expectedTags(AUTHORIZER_CLASS_NAME_CONFIG, "StandardAuthorizer", Map.of("role", "broker")));
+
+ assertMetrics(
+ clusterInstance.controllers().get(controllerId(clusterInstance.type())).metrics(),
+ 4,
+ expectedTags(AUTHORIZER_CLASS_NAME_CONFIG, "StandardAuthorizer", Map.of("role", "controller")));
+ }
+
+ private void assertReplicaSelectorMetrics(ClusterInstance clusterInstance) {
+ assertMetrics(
+ clusterInstance.brokers().get(0).metrics(),
+ MonitorableReplicaSelector.METRICS_COUNT,
+ expectedTags(REPLICA_SELECTOR_CLASS_CONFIG, MonitorableReplicaSelector.class.getSimpleName()));
+ }
+
+ private void assertMetrics(Metrics metrics, int expected, Map<String, String> expectedTags) {
+ int found = 0;
+ for (MetricName metricName : metrics.metrics().keySet()) {
+ if (metricName.group().equals("plugins")) {
+ Map<String, String> tags = metricName.tags();
+ if (expectedTags.equals(tags)) {
+ found++;
+ }
+ }
+ }
+ assertEquals(expected, found);
+ }
+
+ public static class MonitorableReplicaSelector extends RackAwareReplicaSelector implements Monitorable {
+
+ private static final int METRICS_COUNT = 1;
Review Comment:
I've put it in the class as it's the number of metrics this class registers.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2044183839
##########
metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizer.java:
##########
@@ -207,4 +216,45 @@ static AuthorizationResult getDefaultResult(Map<String, ?> configs) {
if (configValue == null) return DENIED;
return Boolean.parseBoolean(configValue.toString().trim()) ? ALLOWED : DENIED;
}
+
+ @Override
+ public void withPluginMetrics(PluginMetrics metrics) {
+ this.authorizerMetrics = new AuthorizerMetrics(metrics);
+ }
+
+ private class AuthorizerMetrics {
Review Comment:
Frankly, no I didn't. This would require updating KIP-877 if we wanted to do so.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2044367536
##########
server/src/test/java/org/apache/kafka/server/MonitorablePluginsIntegrationTest.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server;
+
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Measurable;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Monitorable;
+import org.apache.kafka.common.metrics.PluginMetrics;
+import org.apache.kafka.common.replica.RackAwareReplicaSelector;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.metadata.authorizer.StandardAuthorizer;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.apache.kafka.server.config.ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG;
+import static org.apache.kafka.server.config.ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class MonitorablePluginsIntegrationTest {
+
+ private static int controllerId(Type type) {
+ return type == Type.KRAFT ? 3000 : 0;
+ }
+
+ private static Map<String, String> expectedTags(String config, String clazz) {
+ return expectedTags(config, clazz, Map.of());
+ }
+
+ private static Map<String, String> expectedTags(String config, String clazz, Map<String, String> extraTags) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("config", config);
+ tags.put("class", clazz);
+ tags.putAll(extraTags);
+ return tags;
+ }
+
+ @ClusterTest(
+ types = {Type.KRAFT, Type.CO_KRAFT},
+ serverProperties = {
+ @ClusterConfigProperty(key = StandardAuthorizer.SUPER_USERS_CONFIG, value = "User:ANONYMOUS"),
+ @ClusterConfigProperty(key = AUTHORIZER_CLASS_NAME_CONFIG, value = "org.apache.kafka.metadata.authorizer.StandardAuthorizer"),
+ @ClusterConfigProperty(key = REPLICA_SELECTOR_CLASS_CONFIG, value = "org.apache.kafka.server.MonitorablePluginsIntegrationTest$MonitorableReplicaSelector")
+ }
+ )
+ public void testMonitorableServerPlugins(ClusterInstance clusterInstance) {
+ assertAuthorizerMetrics(clusterInstance);
+ assertReplicaSelectorMetrics(clusterInstance);
+ }
+
+ private void assertAuthorizerMetrics(ClusterInstance clusterInstance) {
+ assertMetrics(
+ clusterInstance.brokers().get(0).metrics(),
+ 4,
+ expectedTags(AUTHORIZER_CLASS_NAME_CONFIG, "StandardAuthorizer", Map.of("role", "broker")));
+
+ assertMetrics(
+ clusterInstance.controllers().get(controllerId(clusterInstance.type())).metrics(),
+ 4,
+ expectedTags(AUTHORIZER_CLASS_NAME_CONFIG, "StandardAuthorizer", Map.of("role", "controller")));
+ }
+
+ private void assertReplicaSelectorMetrics(ClusterInstance clusterInstance) {
+ assertMetrics(
+ clusterInstance.brokers().get(0).metrics(),
+ MonitorableReplicaSelector.METRICS_COUNT,
+ expectedTags(REPLICA_SELECTOR_CLASS_CONFIG, MonitorableReplicaSelector.class.getSimpleName()));
+ }
+
+ private void assertMetrics(Metrics metrics, int expected, Map<String, String> expectedTags) {
+ int found = 0;
+ for (MetricName metricName : metrics.metrics().keySet()) {
+ if (metricName.group().equals("plugins")) {
+ Map<String, String> tags = metricName.tags();
+ if (expectedTags.equals(tags)) {
+ found++;
+ }
+ }
+ }
+ assertEquals(expected, found);
+ }
+
+ public static class MonitorableReplicaSelector extends RackAwareReplicaSelector implements Monitorable {
+
+ private static final int METRICS_COUNT = 1;
Review Comment:
Excuse me, why to set a static constant in a inner class? Maybe it can be moved to `MonitorablePluginsIntegrationTest`?
##########
metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizer.java:
##########
@@ -207,4 +216,45 @@ static AuthorizationResult getDefaultResult(Map<String, ?> configs) {
if (configValue == null) return DENIED;
return Boolean.parseBoolean(configValue.toString().trim()) ? ALLOWED : DENIED;
}
+
+ @Override
+ public void withPluginMetrics(PluginMetrics metrics) {
+ this.authorizerMetrics = new AuthorizerMetrics(metrics);
+ }
+
+ private class AuthorizerMetrics {
+ private final Sensor authorizationAllowedSensor;
+ private final Sensor authorizationDeniedSensor;
+ private final Sensor authorizationRequestSensor;
+
+ private AuthorizerMetrics(PluginMetrics metrics) {
+ authorizationAllowedSensor = metrics.addSensor("authorizer-authorization-allowed");
Review Comment:
those metrics are bound with `StandardAuthorizer`, and maybe the metrics in KIP-877 should use the explicit authorizer name.
```
kafka.server:type=plugins,config=authorizer.class.name,class=StandardAuthorizer,name=acls-total-count
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2041145949
##########
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java:
##########
@@ -40,14 +47,46 @@ private Plugin(T instance, PluginMetricsImpl pluginMetrics) {
this.pluginMetrics = Optional.ofNullable(pluginMetrics);
}
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param tagsSupplier supplier to retrieve the tags
+ * @return the plugin
+ */
+ public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, Supplier<Map<String, String>> tagsSupplier) {
+ PluginMetricsImpl pluginMetrics = null;
+ if (instance instanceof Monitorable && metrics != null) {
+ pluginMetrics = new PluginMetricsImpl(metrics, tagsSupplier.get());
+ ((Monitorable) instance).withPluginMetrics(pluginMetrics);
+ }
+ return new Plugin<>(instance, pluginMetrics);
+ }
Review Comment:
I agree. I’d prefer to require a `LinkedHashMap` rather than just mentioning it in the JavaDoc, since JavaDoc alone doesn’t enforce anything.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2041168550
##########
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java:
##########
@@ -40,14 +47,46 @@ private Plugin(T instance, PluginMetricsImpl pluginMetrics) {
this.pluginMetrics = Optional.ofNullable(pluginMetrics);
}
+ /**
+ * Wrap an instance into a Plugin.
+ * @param instance the instance to wrap
+ * @param metrics the metrics
+ * @param tagsSupplier supplier to retrieve the tags
+ * @return the plugin
+ */
+ public static <T> Plugin<T> wrapInstance(T instance, Metrics metrics, Supplier<Map<String, String>> tagsSupplier) {
+ PluginMetricsImpl pluginMetrics = null;
+ if (instance instanceof Monitorable && metrics != null) {
+ pluginMetrics = new PluginMetricsImpl(metrics, tagsSupplier.get());
+ ((Monitorable) instance).withPluginMetrics(pluginMetrics);
+ }
+ return new Plugin<>(instance, pluginMetrics);
+ }
Review Comment:
Feel free to open a jira if you want to handle this.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18888: Add KIP-877 support to Authorizer [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19050:
URL: https://github.com/apache/kafka/pull/19050#discussion_r2031317020
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3284,7 +3285,7 @@ private boolean maybeUpdateRegularExpressions(
* @param log The log instance.
* @param time The time instance.
* @param image The metadata image to use for listing the topics.
- * @param authorizer The authorizer.
+ * @param authorizerPlugin The authorizer.
Review Comment:
How about update to `The authorizer is wrapped in a {@link org.apache.kafka.common.internals.Plugin}`?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3352,15 +3353,15 @@ public static Map<String, ResolvedRegularExpression> refreshRegularExpressions(
* that the member is authorized to describe.
*
* @param context The request context.
- * @param authorizer The authorizer.
+ * @param authorizerPlugin The authorizer.
Review Comment:
ditto: How about update to `The authorizer is wrapped in a {@link org.apache.kafka.common.internals.Plugin}`?
##########
server-common/src/main/java/org/apache/kafka/server/network/EndpointReadyFutures.java:
##########
@@ -79,17 +80,17 @@ public Builder addReadinessFutures(
/**
* Build the EndpointReadyFutures object.
*
- * @param authorizer The authorizer to use, if any. Will be started.
- * @param info Server information to be passed to the authorizer.
+ * @param authorizerPlugin The authorizer to use, if any. Will be started.
Review Comment:
How about update to `The authorizer which is wrapped in a {@link org.apache.kafka.common.internals.Plugin} to use`?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org