KAFKA-18888: Add KIP-877 support to Authorizer (#19050)

This also adds metrics to StandardAuthorizer

Reviewers: Chia-Ping Tsai <[email protected]>, Ken Huang
 <[email protected]>, Jhen-Yung Hsu <[email protected]>, TaiJuWu
 <[email protected]>

Author: mimaison

Diff for: checkstyle/import-control-metadata.xml

@@ -176,6 +176,9 @@
             <allow pkg="org.apache.kafka.controller" />
             <allow pkg="org.apache.kafka.metadata" />
             <allow pkg="org.apache.kafka.common.internals" />
+            <allow pkg="org.apache.kafka.common.metrics" />
+            <allow pkg="org.apache.kafka.common.metrics.internals" />
+            <allow pkg="org.apache.kafka.common.metrics.stats" />
         </subpackage>
         <subpackage name="bootstrap">
             <allow pkg="org.apache.kafka.snapshot" />

Diff for: checkstyle/import-control-server.xml

@@ -105,6 +105,7 @@
   <subpackage name="security">
     <allow pkg="org.apache.kafka.common.resource" />
     <allow pkg="org.apache.kafka.network" />
+    <allow pkg="org.apache.kafka.server" />
     <allow pkg="org.apache.kafka.server.authorizer" />
   </subpackage>
 

Diff for: clients/src/main/java/org/apache/kafka/common/internals/Plugin.java

@@ -30,6 +30,13 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
 
+/**
+ * Plugins have the following tags:
+ * <ul>
+ *   <li><code>config</code> set to the name of the configuration to specifying the plugin</li>
+ *   <li><code>class</code> set to the name of the instance class</li>
+ * </ul>
+ */
 public class Plugin<T> implements Supplier<T>, AutoCloseable {
 
     private final T instance;
@@ -40,14 +47,49 @@ 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) {
-        Map<String, String> tags = tags(key, instance);
-        tags.putAll(extraTags);
-        return wrapInstance(instance, metrics, () -> tags);
+    /**
+     * 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) {
+        Supplier<Map<String, String>> tagsSupplier = () -> {
+            Map<String, String> tags = tags(key, instance);
+            tags.put(name, value);
+            return tags;
+        };
+        return wrapInstance(instance, metrics, tagsSupplier);
     }
 
     private static <T> Map<String, String> tags(String key, T instance) {
@@ -57,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
+     * @param metrics the metrics
+     * @param key the value for the <code>config</code> tag
+     * @return the list of plugins
+     */
     public static <T> List<Plugin<T>> wrapInstances(List<T> instances, Metrics metrics, String key) {
         List<Plugin<T>> plugins = new ArrayList<>();
         for (T instance : instances) {
@@ -65,15 +114,6 @@ public static <T> List<Plugin<T>> wrapInstances(List<T> instances, Metrics metri
         return plugins;
     }
 
-    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);
-    }
-
     @Override
     public T get() {
         return instance;

Diff for: clients/src/main/java/org/apache/kafka/server/authorizer/Authorizer.java

@@ -59,6 +59,10 @@
  *
  * Authorizer implementation class may optionally implement @{@link org.apache.kafka.common.Reconfigurable}
  * to enable dynamic reconfiguration without restarting the broker.
+ * <p>Authorizer implementation class may also optionally implement {@link org.apache.kafka.common.metrics.Monitorable}
+ * to enable the authorizer to register metrics. The following tags are automatically added to all metrics registered:
+ * <code>config</code> set to <code>authorizer.class.name</code>, <code>class</code> set to the Authorizer class name,
+ * and <code>role</code> set to either <code>broker</code> or <code>controller</code>.
  * <p>
  * <b>Threading model:</b>
  * <ul>

Diff for: core/src/main/java/kafka/server/QuotaFactory.java

@@ -27,7 +27,6 @@
 import org.apache.kafka.server.quota.ClientQuotaCallback;
 import org.apache.kafka.server.quota.QuotaType;
 
-import java.util.Map;
 import java.util.Optional;
 
 import scala.Option;
@@ -150,7 +149,7 @@ private static Optional<Plugin<ClientQuotaCallback>> createClientQuotaCallback(
             clientQuotaCallback,
             metrics,
             QuotaConfig.CLIENT_QUOTA_CALLBACK_CLASS_CONFIG,
-            Map.of("role", role)
+            "role", role
         ));
     }
 

Diff for: core/src/main/java/kafka/server/builders/KafkaApisBuilder.java

@@ -28,6 +28,7 @@
 import kafka.server.ReplicaManager;
 import kafka.server.share.SharePartitionManager;
 
+import org.apache.kafka.common.internals.Plugin;
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.coordinator.group.GroupConfigManager;
@@ -58,7 +59,7 @@ public class KafkaApisBuilder {
     private ConfigRepository configRepository = null;
     private MetadataCache metadataCache = null;
     private Metrics metrics = null;
-    private Optional<Authorizer> authorizer = Optional.empty();
+    private Optional<Plugin<Authorizer>> authorizerPlugin = Optional.empty();
     private QuotaManagers quotas = null;
     private FetchManager fetchManager = null;
     private SharePartitionManager sharePartitionManager = null;
@@ -131,8 +132,8 @@ public KafkaApisBuilder setMetrics(Metrics metrics) {
         return this;
     }
 
-    public KafkaApisBuilder setAuthorizer(Optional<Authorizer> authorizer) {
-        this.authorizer = authorizer;
+    public KafkaApisBuilder setAuthorizerPlugin(Optional<Plugin<Authorizer>> authorizerPlugin) {
+        this.authorizerPlugin = authorizerPlugin;
         return this;
     }
 
@@ -219,7 +220,7 @@ public KafkaApis build() {
                              configRepository,
                              metadataCache,
                              metrics,
-                             OptionConverters.toScala(authorizer),
+                             OptionConverters.toScala(authorizerPlugin),
                              quotas,
                              fetchManager,
                              sharePartitionManager,

Diff for: core/src/main/scala/kafka/server/AclApis.scala

@@ -22,6 +22,7 @@ import kafka.utils.Logging
 import org.apache.kafka.common.acl.AclOperation._
 import org.apache.kafka.common.acl.AclBinding
 import org.apache.kafka.common.errors._
+import org.apache.kafka.common.internals.Plugin
 import org.apache.kafka.common.message.CreateAclsResponseData.AclCreationResult
 import org.apache.kafka.common.message.DeleteAclsResponseData.DeleteAclsFilterResult
 import org.apache.kafka.common.message._
@@ -46,7 +47,7 @@ import scala.jdk.OptionConverters.RichOptional
  * Logic to handle ACL requests.
  */
 class AclApis(authHelper: AuthHelper,
-              authorizer: Option[Authorizer],
+              authorizerPlugin: Option[Plugin[Authorizer]],
               requestHelper: RequestHandlerHelper,
               role: ProcessRole,
               config: KafkaConfig) extends Logging {
@@ -61,7 +62,7 @@ class AclApis(authHelper: AuthHelper,
   def handleDescribeAcls(request: RequestChannel.Request): CompletableFuture[Unit] = {
     authHelper.authorizeClusterOperation(request, DESCRIBE)
     val describeAclsRequest = request.body[DescribeAclsRequest]
-    authorizer match {
+    authorizerPlugin match {
       case None =>
         requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
           new DescribeAclsResponse(new DescribeAclsResponseData()
@@ -74,7 +75,7 @@ class AclApis(authHelper: AuthHelper,
         requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
           new DescribeAclsResponse(new DescribeAclsResponseData()
             .setThrottleTimeMs(requestThrottleMs)
-            .setResources(DescribeAclsResponse.aclsResources(auth.acls(filter))),
+            .setResources(DescribeAclsResponse.aclsResources(auth.get.acls(filter))),
           describeAclsRequest.version))
     }
     CompletableFuture.completedFuture[Unit](())
@@ -84,7 +85,7 @@ class AclApis(authHelper: AuthHelper,
     authHelper.authorizeClusterOperation(request, ALTER)
     val createAclsRequest = request.body[CreateAclsRequest]
 
-    authorizer match {
+    authorizerPlugin match {
       case None => requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
         createAclsRequest.getErrorResponse(requestThrottleMs,
           new SecurityDisabledException("No Authorizer is configured.")))
@@ -109,7 +110,7 @@ class AclApis(authHelper: AuthHelper,
         }
 
         val future = new CompletableFuture[util.List[AclCreationResult]]()
-        val createResults = auth.createAcls(request.context, validBindings.asJava).stream().map(_.toCompletableFuture).toList
+        val createResults = auth.get.createAcls(request.context, validBindings.asJava).stream().map(_.toCompletableFuture).toList
 
         def sendResponseCallback(): Unit = {
           val aclCreationResults = allBindings.map { acl =>
@@ -139,7 +140,7 @@ class AclApis(authHelper: AuthHelper,
   def handleDeleteAcls(request: RequestChannel.Request): CompletableFuture[Unit] = {
     authHelper.authorizeClusterOperation(request, ALTER)
     val deleteAclsRequest = request.body[DeleteAclsRequest]
-    authorizer match {
+    authorizerPlugin match {
       case None =>
         requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
           deleteAclsRequest.getErrorResponse(requestThrottleMs,
@@ -148,7 +149,7 @@ class AclApis(authHelper: AuthHelper,
       case Some(auth) =>
 
         val future = new CompletableFuture[util.List[DeleteAclsFilterResult]]()
-        val deleteResults = auth.deleteAcls(request.context, deleteAclsRequest.filters)
+        val deleteResults = auth.get.deleteAcls(request.context, deleteAclsRequest.filters)
           .stream().map(_.toCompletableFuture).toList
 
         def sendResponseCallback(): Unit = {

Diff for: core/src/main/scala/kafka/server/AuthHelper.scala

@@ -24,6 +24,7 @@ import org.apache.kafka.clients.admin.EndpointType
 import org.apache.kafka.common.acl.AclOperation
 import org.apache.kafka.common.acl.AclOperation.DESCRIBE
 import org.apache.kafka.common.errors.ClusterAuthorizationException
+import org.apache.kafka.common.internals.Plugin
 import org.apache.kafka.common.message.DescribeClusterResponseData
 import org.apache.kafka.common.message.DescribeClusterResponseData.DescribeClusterBrokerCollection
 import org.apache.kafka.common.protocol.Errors
@@ -38,7 +39,7 @@ import org.apache.kafka.server.authorizer.{Action, AuthorizationResult, Authoriz
 import scala.collection.Seq
 import scala.jdk.CollectionConverters._
 
-class AuthHelper(authorizer: Option[Authorizer]) {
+class AuthHelper(authorizer: Option[Plugin[Authorizer]]) {
   def authorize(requestContext: RequestContext,
                 operation: AclOperation,
                 resourceType: ResourceType,
@@ -49,7 +50,7 @@ class AuthHelper(authorizer: Option[Authorizer]) {
     authorizer.forall { authZ =>
       val resource = new ResourcePattern(resourceType, resourceName, PatternType.LITERAL)
       val actions = Collections.singletonList(new Action(operation, resource, refCount, logIfAllowed, logIfDenied))
-      authZ.authorize(requestContext, actions).get(0) == AuthorizationResult.ALLOWED
+      authZ.get.authorize(requestContext, actions).get(0) == AuthorizationResult.ALLOWED
     }
   }
 
@@ -64,7 +65,7 @@ class AuthHelper(authorizer: Option[Authorizer]) {
       case Some(authZ) =>
         val resourcePattern = new ResourcePattern(resource.resourceType, resource.name, PatternType.LITERAL)
         val actions = supportedOps.map { op => new Action(op, resourcePattern, 1, false, false) }
-        authZ.authorize(request.context, actions.asJava).asScala
+        authZ.get.authorize(request.context, actions.asJava).asScala
           .zip(supportedOps)
           .filter(_._1 == AuthorizationResult.ALLOWED)
           .map(_._2).toSet
@@ -77,7 +78,7 @@ class AuthHelper(authorizer: Option[Authorizer]) {
   def authorizeByResourceType(requestContext: RequestContext, operation: AclOperation,
                               resourceType: ResourceType): Boolean = {
     authorizer.forall { authZ =>
-      authZ.authorizeByResourceType(requestContext, operation, resourceType) == AuthorizationResult.ALLOWED
+      authZ.get.authorizeByResourceType(requestContext, operation, resourceType) == AuthorizationResult.ALLOWED
     }
   }
 
@@ -109,7 +110,7 @@ class AuthHelper(authorizer: Option[Authorizer]) {
           val resource = new ResourcePattern(resourceType, resourceName, PatternType.LITERAL)
           new Action(operation, resource, count, logIfAllowed, logIfDenied)
         }.toBuffer
-        authZ.authorize(requestContext, actions.asJava).asScala
+        authZ.get.authorize(requestContext, actions.asJava).asScala
           .zip(resourceNameToCount.keySet)
           .collect { case (authzResult, resourceName) if authzResult == AuthorizationResult.ALLOWED =>
             resourceName

Diff for: core/src/main/scala/kafka/server/BrokerServer.scala

@@ -27,6 +27,7 @@ import kafka.server.metadata._
 import kafka.server.share.{ShareCoordinatorMetadataCacheHelperImpl, SharePartitionManager}
 import kafka.utils.CoreUtils
 import org.apache.kafka.common.config.ConfigException
+import org.apache.kafka.common.internals.Plugin
 import org.apache.kafka.common.message.ApiMessageType.ListenerType
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.network.ListenerName
@@ -103,7 +104,7 @@ class BrokerServer(
 
   @volatile var dataPlaneRequestProcessor: KafkaApis = _
 
-  var authorizer: Option[Authorizer] = None
+  var authorizerPlugin: Option[Plugin[Authorizer]] = None
   @volatile var socketServer: SocketServer = _
   var dataPlaneRequestHandlerPool: KafkaRequestHandlerPool = _
 
@@ -412,8 +413,7 @@ class BrokerServer(
       )
 
       // Create and initialize an authorizer if one is configured.
-      authorizer = config.createNewAuthorizer()
-      authorizer.foreach(_.configure(config.originals))
+      authorizerPlugin = config.createNewAuthorizer(metrics, ProcessRole.BrokerRole.toString)
 
       // The FetchSessionCache is divided into config.numIoThreads shards, each responsible
       // for Math.max(1, shardNum * sessionIdRange) <= sessionId < (shardNum + 1) * sessionIdRange
@@ -456,7 +456,7 @@ class BrokerServer(
         configRepository = metadataCache,
         metadataCache = metadataCache,
         metrics = metrics,
-        authorizer = authorizer,
+        authorizerPlugin = authorizerPlugin,
         quotas = quotaManagers,
         fetchManager = fetchManager,
         sharePartitionManager = sharePartitionManager,
@@ -529,7 +529,7 @@ class BrokerServer(
           config.nodeId,
           sharedServer.metadataPublishingFaultHandler,
           "broker",
-          authorizer.toJava
+          authorizerPlugin.toJava
         ),
         sharedServer.initialBrokerMetadataLoadFaultHandler,
         sharedServer.metadataPublishingFaultHandler
@@ -586,7 +586,7 @@ class BrokerServer(
       // authorizer future is completed.
       val endpointReadyFutures = {
         val builder = new EndpointReadyFutures.Builder()
-        builder.build(authorizer.toJava,
+        builder.build(authorizerPlugin.toJava,
           new KafkaAuthorizerServerInfo(
             new ClusterResource(clusterId),
             config.nodeId,
@@ -645,7 +645,7 @@ class BrokerServer(
       .withGroupCoordinatorMetrics(new GroupCoordinatorMetrics(KafkaYammerMetrics.defaultRegistry, metrics))
       .withGroupConfigManager(groupConfigManager)
       .withPersister(persister)
-      .withAuthorizer(authorizer.toJava)
+      .withAuthorizerPlugin(authorizerPlugin.toJava)
       .build()
   }
 
@@ -765,7 +765,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 plugin"))
 
       /**
        * We must shutdown the scheduler early because otherwise, the scheduler could touch other