...

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 mode 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. Authorization mode 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 out-of-the-box authorizer class SimpleAclAuthorizer to implement 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 implementing org.apache.kafka.common.ClusterResourceListener will will also be provided cluster id which may be included in logs or used to support centralized ACL storage.

Public Interfaces

Authorizer Configuration

A new The existing configuration option `authorizer.class` .name` will be introduced extended to configure a support broker authorizer authorizers using the new Java interface org.apache.kafka.server.authorizer.Authorizer . The existing config `authorizer.class.name` will be deprecated, but will continue to support authorizers using the existing Scala trait kafka.security.auth.Authorizer.New configuration option:

• Name: authorizer.class
• Type: CLASS

Authorizer API

The new Java Authorizer interface and its supporting classes are shown below:

package org.apache.kafka.server.authorizer;

import java.io.Closeable;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import org.apache.kafka.common.ClusterResource;
import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.KafkaRequestContext;

/**
 *
 * Pluggable authorizer interface for Kafka brokers.
 *
 * Startup sequence in brokers:
 * <ol>
 *   <li>Broker creates authorizer instance if configured in `authorizer.class` or `authorizer.class.name`.</li>
 *   <li>Broker configures and starts authorizer instance. Authorizer implementation starts loading its metadata.</li>
 *   <li>Broker starts SocketServer to accept connections and process requests.</li>
 *   <li>For each listener, SocketServer waits for authorization metadata to be available in the
 *       authorizer before accepting connections. The future returned by {@link #start(ClusterResource, Map)} )}
 *       must return only when authorizer is ready to authorize requests on the listener.</li>
 *   <li>Broker accepts connections. For each connection, broker performs authentication and then accepts Kafka requests.
 *       For each request, broker invokes {@link #authorize(KafkaRequestContext, List)} to authorize
 *       actions performed by the request.</li>
 * </ol>
 *
 * Authorizer implementation class may optionally implement @{@link org.apache.kafka.common.Reconfigurable}
 * to enable dynamic reconfiguration without restarting the broker.
 * <p>
 * <b>Thread safety:</b> All authorizer operations including authorization and ACL updates must be thread-safe.
 * </p>
 */
@InterfaceStability.Evolving
public interface Authorizer extends Configurable, Closeable {

    /**
     * Starts loading authorization metadata and returns futures that can be used to wait until
     * metadata for authorizing requests on each listener is available. Each listener will be
     * started only after its metadata is available and authorizer is ready to start authorizing
     * requests on that listener.
     *
     * @param clusterResource Cluster metadata for the Kafka cluster
     * @param listeners Listener names with their security protocols
     * @return CompletableFutures for each listener that completes when authorizer is ready to
     *         start authorizing requests on that listener. Returned map contains one future
     *         for each listener name in the input `listeners` map.
     */
    Map<String, CompletableFuture<Void>> start(ClusterResource clusterResource,
                                               Map<String, SecurityProtocol> listeners);

    /**
     * Authorizes the specified action. Additional metadata for the action is specified
     * in `requestContext`.
     *
     * @param requestContext Request context including request type, security protocol and listener name
     * @param actions Actions being authorized including resource and operation for each action
     * @return List of authorization results for each action in the same order as the provided actions
     */
    List<AuthorizationResult> authorize(KafkaRequestContext requestContext, List<Action> actions);

    /**
     * Creates new ACL bindings.
     *
     * @param requestContext Request context if the ACL is being created by a broker to handle
     *        a client request to create ACLs. This may be null if ACLs are created directly in ZooKeeper
     *        using AclCommand.
     * @param aclBindings ACL bindings to create
     *
     * @return Create result for each ACL binding in the same order as in the input list
     */
    List<AclCreateResult> createAcls(KafkaRequestContext requestContext, List<AclBinding> aclBindings);

    /**
     * Deletes all ACL bindings that match the provided filters.
     *
     * @param requestContext Request context if the ACL is being deleted by a broker to handle
     *        a client request to delete ACLs. This may be null if ACLs are deleted directly in ZooKeeper
     *        using AclCommand.
     * @param aclBindingFilters Filters to match ACL bindings that are to be deleted
     *
     * @return Delete result for each filter in the same order as in the input list.
     *         Each result indicates which ACL bindings were actually deleted as well as any
     *         bindings that matched but could not be deleted.
     */
    List<AclDeleteResult> deleteAcls(KafkaRequestContext requestContext, List<AclBindingFilter> aclBindingFilters);

    /**
     * Returns ACL bindings which match the provided filter.
     *
     * @return Iterator for ACL bindings, which may be populated lazily.
     */
    Iterable<AclBinding> acls(AclBindingFilter filter);
}

...

package org.apache.kafka.server.authorizer;

import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.resource.PatternType;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourceType;

@InterfaceStability.Evolving
public class Action {

    private final ResourcePattern resourcePattern;
    private final AclOperation operation;
    private final AuthorizationMode authorizationMode;
    private final int count;

    public Action(AclOperation operation,
                  ResourceType resourceType,
                  String resourceName,
                  AuthorizationMode authorizationMode,
                  int count) {
        this.operation = operation;
        this.resourcePattern = new ResourcePattern(resourceType, resourceName, PatternType.LITERAL);
        this.authorizationMode = authorizationMode;
        this.count = count;
    }

    /**
     * Resource on which action is being performed.
     */
    public ResourcePattern resourcePattern() {
        return resourcePattern;
    }

    /**
     * Operation being performed.
     */
    public AclOperation operation() {
        return operation;
    }

    /**
     * Authorization mode to enable authorization logs to distinguish between attempts
     * to access unauthorized resources and other filtering operations performed by the broker.
     */
    public AuthorizationMode authorizationMode() {
        return authorizationMode;
    }

    /**
     * Number of times the authorization result is used. For example, a single topic
     * authorization result may be used with `count` partitions of the topic within a single
     * request.
     */
    public int count() {
        return count;
    }

Authorization Mode provides additional context for audit logging:

package org.apache.kafka.server.authorizer;

public enum AuthorizationMode {
    /**
     * Access was requested to resource. If authorization result is ALLOWED, access is granted to
     * the resource to perform the request. If DENIEDDENY, request is failed with authorization failure.
     */ <p>
     MANDATORY,

* Audit logs tracking /**
ALLOWED access should include this * Access was requested to resource. If authorization if result is ALLOWED, access is granted to.
     * theAudit resourcelogs totracking performDENIED theaccess request.should Ifinclude DENIED,this alternativeif authorizationresult rules are appliedis DENIED.
     * to determine if access is allowed.</p>
     */
    OPTIONALMANDATORY,

    /**
     * Access was requested to authorized resources (e.g. to subscribe to regex pattern).resource. If authorization result is ALLOWED, access is granted to
     * the resource to perform the request. If DENY, alternative authorization rules are applied
     * Request is performed on resources whose authorization result is ALLOWED and the rest of to determine if access is allowed.
     * <p>
     * For example, topic creation is allowed if user has Cluster:Create
     * permission to create any topic or the fine-grained Topic:Create permission to create topics
     * of the requested name. Cluster:Create is an optional ACL in this case.
     * </p><p>
     * Audit logs tracking ALLOWED access should include this if result is ALLOWED.
     * Audit logs tracking DENIED access can omit this if result is DENIED, since an alternative
     * authorization is used to determine access.
     * </p>
     */
    OPTIONAL,

    /**
     * Access was requested to authorized resources (e.g. to subscribe to regex pattern).
     * Request is performed on resources whose authorization result is ALLOWED and the rest of
     * the resources are filtered out.
     * <p>
     * Audit logs tracking ALLOWED access should include this if result is ALLOWED.
     * Audit logs tracking DENIED access can omit this if result is DENIED since access was not
     * actually requested for the specified resource and it is filtered out.
     * </p>
     */
    FILTER,

    /**
     * Request to list authorized operations. No access is actually performed by this request
     * thebased resourceson arethe filteredauthorization outresult.
     */ <p>
     FILTER,

* Audit logs tracking /**
     * Request to list authorized operations. No access is actually performed by this requestALLOWED/DENIED access can omit these since no access is performed
     * as a result of this.
     * based on the authorization result.</p>
     */
    LIST_AUTHORIZED
}

Authorize method returns individual allowed/denied results for every action. ACL create and delete operations will return any exceptions from each access control entry requested.

...

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 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 used.

Optional Interfaces

ClusterResourceListener

...

Reconfigurable

kafka.server.DynamicBrokerConfig will be updated to support dynamic update of authorizers which implement org.apache.kafka.common.Reconfigurable. Authorizer implementations can react to dynamic updates of any of its configs including custom configs, avoiding broker restarts to update configs.

...

Existing authorizer interfaces and classes are being deprecated, but not removed. We will continue to support the old same config with the old API to ensure that existing users are not impacted.

If we are changing behavior how will we phase out the older behavior?

We are deprecating the existing config `authorizer.class.name` and the existing Scala authorizer API. These will be removed in a future release, but will continue to be supported for backward compatibility until then. Until the old authorizer is removed, no config changes are required during upgrade.

...