...

1. Define a new Java interface for authorizer in 'clients' module in the package 'org.apache.kafka.server' similar to other server-side pluggable classes.
2. KIP-4 has added ACL-related classes in the package org.apache.kafka.common (e.g. ResourcePattern and AccessControlEntry) to support ACL management using AdminClient. We will attempt to reuse these classes wherever possible.
3. Deprecate but retain existing Scala authorizer API for backward compatibility to ensure that existing custom authorizers can be used with new brokers.
4. Provide context about the request to authorizers to enable context-specific logic based on security protocol or listener to be applied to authorization.
5. Provide additional context about the request including ApiKey and correlation id from the request header since these are useful for matching debug-level authorization logs with corresponding request logs.
6. For ACL updates, provide request context including principal requesting the update and the listener on which request arrived to enable additional validation.
7. Return individual responses for each access control entry update when multiple entries of a resource are updated. At the moment, we update the ZooKeeper node for a resource pattern multiple times when a request adds or removes multiple entries for a resource in a single update request. Since it is a common usage pattern to add or remove multiple access control entries while updating ACLs for a resource, batched updates will be supported to enable a single atomic update for each resource pattern.
8. Provide authorization usage flag to authorizers to enable authorization logs to indicate attempts to access unauthorized resources. Kafka brokers log denied operations at INFO level and allowed operations at DEBUG level with the expectation that denied operations are rare and indicate erroneous or malicious use of the system. But we currently have several uses of Authorizer#authorize for filtering accessible resources or operations, for example for regex subscription. These fill up authorization logs with denied log entries, making these logs unusable for determining actual attempts to access resources by users who don’t have appropriate permissions. Audit flag will enable the authorizer to determine the severity of denied access.
9. For authorizers that don’t store metadata in ZooKeeper, ensure that authorizer metadata for each listener is available before starting up the listener. This enables different authorization metadata stores for different listeners.
10. Rewrite Add a new out-of-the-box authorizer class SimpleAclAuthorizer to implement that implements the new authorizer interface, making use of the features supported by the new API.
11. Retain existing audit log entry format in SimpleAclAuthorizer to ensure that tools that parse these logs continue to work.
12. Enable Authorizer implementations to make use of additional Kafka interfaces similar to other pluggable callbacks. Authorizers can implement org.apache.kafka.common.Reconfigurable  to support dynamic reconfiguration without restarting the broker. Authorizers will also be provided cluster id which may be included in logs or used to support centralized ACL storage.

...

package org.apache.kafka.server.authorizer;

import java.util.Collections;
import java.util.Collection;
import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.errors.ApiException;

@InterfaceStability.Evolving
public class AclDeleteResult {
    private final ApiException exception;
    private final Collection<AclBindingDeleteResult> aclBindingDeleteResults;

    public AclDeleteResult(ApiException exception) {
        this(Collections.emptySet(), exception);
    }

    public AclDeleteResult(Collection<AclBindingDeleteResult> deleteResults) {
        this(deleteResults, null);
    }

    private AclDeleteResult(Collection<AclBindingDeleteResult> deleteResults, ApiException exception) {
        this.aclBindingDeleteResults = deleteResults;
        this.exception = exception;
    }

    /**
     * Returns any exception while attempting to match ACL filter to delete ACLs.
     */
    public ApiException exception() {
        return exception;
    }

    /**
     * Returns delete result for each matching ACL binding.
     */
    public Collection<AclBindingDeleteResult> aclBindingDeleteResults() {
        return aclBindingDeleteResults;
    }


    /**
     * Delete result for each ACL binding that matched a delete filter.
     */
    public static class AclBindingDeleteResult {
        private final AclBinding aclBinding;
        private final ApiException exception;

        public AclBindingDeleteResult(AclBinding aclBinding) {
            this(aclBinding, null);
        }

        public AclBindingDeleteResult(AclBinding aclBinding, ApiException exception) {
            this.aclBinding = aclBinding;
            this.exception = exception;
        }

        /**
         * Returns ACL binding that matched the delete filter. {@link #deleted()} indicates if
         * the binding was deleted.
         */
        public AclBinding aclBinding() {
            return aclBinding;
        }

        /**
         * Returns exception that resulted in failure to delete ACL binding.
         */
        public ApiException exception() {
            return exception;
        }

        /**
         * Returns true if ACL binding was deleted, false otherwise.
         */
        public boolean deleted() {
            return exception == null;
        }
    }
}

Proposed Changes

Deprecate existing Scala Authorizer Trait

kafka.security.auth.Authorizer will be deprecated along with all the supporting Scala classes including Resource, Operations and ResourceTypes. A new AuthorizerWrapper class will be introduced to wrap implementations using the Scala trait into the new Java interface. All usage of Authorizer (e.g. in KafkaApis) will be replaced with the new authorizer interface.

SimpleAclAuthorizer

AclAuthorizer

A new authorizer implementation will be added. kafka.security.AclAuthorizer will SimpleAclAuthorizer will be updated to implement the new interface, making use of the additional request context available to improve authorization logging. This enables the authorizer to will be used with the new config 'authorizer.class'. SimpleAclAuthorizer will remain a Scala class in `core`.  SimpleAclAuthorizer will continue to implement the old Scala Authorizer trait so that it can continue to be used with the old config 'authorizer.class.name', Internally, it will be handled as the new interface without a wrapper regardless of which config was compatible with SimpleAclAuthorizer and will support all its existing configs including super.users.

SimpleAclAuthorizer

SimpleAclAuthorizer will be deprecated, but we will continue to support this implementation using the old API. Since it is part of the public API, all its public methods will be retained without change. This enables existing custom implementations that rely on this class to continue to be used.

Optional Interfaces

Reconfigurable

...

In some environments, authorization decisions may be dependent on the security protocol used by the client or the listener on which the request was received. We have listener prefixed configs to enable independent listener-specific configs for authentication etc. But since authorizers tend to cache a lot of metadata and need to watch for changes in metadata, a single shared instance works better for authorization. This KIP proposes a single authorizer that can use listener and security protocol provided in the authorization context to include listener-specific authorization logic.

Extend SimpleAclAuthorizer to support the new API

SimpleAclAuthorizer is part of our public API since it is in the public package kafka.security.auth. So we need to ensure that the old API is used with this authorizer if custom implementations extend this class and override specific methods. So this KIP deprecates, but retains this implementation and adds a separate implementation that uses the new API.