Status

Current state: "Under DiscussionAccepted"

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA:

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

...

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 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. Authorization mode 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 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

1. Enable asynchronous ACL updates to avoid blocking broker request threads when ACLs are updated in a remote store (e.g. a database).

Public Interfaces

Authorizer Configuration

The existing configuration A new 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.Set;
import java.util.concurrent.CompletableFutureCompletionStage;
import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.acl.AccessControlEntryEndpoint;
import org.apache.kafka.common.annotationacl.InterfaceStabilityAclBinding;
import org.apache.kafka.common.resourceacl.ResourcePatternAclBindingFilter;
import org.apache.kafka.common.securityannotation.auth.SecurityProtocolInterfaceStability;
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(String, SecurityProtocolAuthorizerServerInfo)}
 *       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(KafkaRequestContextAuthorizableRequestContext, List)} to authorize
 *       actions performed by the request.</li>
 * </ol>
 *
 * Authorizer implementation class may optionally implement the following interfaces:
 * <ul>
 *   <li>@@{@link org.apache.kafka.common.Reconfigurable}
 * to enable dynamic reconfiguration
 *   without restarting the broker.</li>
 * <p>
 * <li>{@link org.apache.kafka.common.ClusterResourceListener} to obtain cluster id</li><b>Threading model:</b>
 * </ul><ul>
 */
@InterfaceStability.Evolving
public interface Authorizer extends<li>All Configurable,authorizer Closeableoperations {

including authorization and ACL /**
updates must be thread-safe.</li>
 * * Starts loading<li>ACL authorizationupdate metadatamethods forare theasynchronous. listenerImplementations andwith returnslow aupdate futurelatency thatmay canreturn bea
     * used to wait until all metadata forcompleted authorizingfuture requestsusing on the specified listener are available.
{@link java.util.concurrent.CompletableFuture#completedFuture(Object)}.
 *       This ensures *that Eachthe listenerrequest will be startedhandled onlysynchronously afterby itsthe metadatacaller iswithout availableusing anda
 authorizer* is ready to
    purgatory *to startwait authorizingfor requests on that listener.
     *
     * @param listenerNamethe result. If ACL updates require remote communication which may block,
 *       return Namea offuture listener
that is completed asynchronously when *the @paramremote securityProtocoloperation completes. SecurityThis protocolenables
 used* by the listener
    the *caller @returnto CompletableFutureprocess thatother completesrequests whenon authorizerthe isrequest readythreads to start authorizingwithout blocking.</li>
 *   <li>Any *threads or thread pools used for processing remote operations requestsasynchronously oncan thebe specifiedstarted listenerduring
 *    */
   {@link CompletableFuture<Void> start(String listenerName, SecurityProtocol securityProtocol);

    /**
     * Authorizes the specified action. Additional metadata for the action is specified
     * in `requestContext`.
     *#start(AuthorizerServerInfo)}. These threads must be shutdown during {@link Authorizer#close()}.</li>
 * </ul>
 * </p>
 */
@InterfaceStability.Evolving
public interface Authorizer extends Configurable, Closeable {

    /**
     * Starts loading authorization metadata and returns futures that can be used to wait until
     * @parammetadata requestContextfor Requestauthorizing contextrequests includingon requesteach type,listener securityis protocolavailable. andEach listener will namebe
     * @paramstarted actionsonly Actionsafter beingits authorizedmetadata includingis resourceavailable and operation for each actionauthorizer is ready to start authorizing
     * @returnrequests Liston ofthat authorizationlistener.
 results for each action in*
 the same order as the* provided@param actions
serverInfo Metadata for the broker */
including broker id and List<AuthorizationResult> authorize(KafkaRequestContext requestContext, List<Action> actions);

    /**
     * Creates a new ACL binding.listener endpoints
     * @return CompletionStage for each endpoint that completes when authorizer is ready to
     *
     * @param requestContext Request contextstart ifauthorizing therequests ACLon isthat beinglistener.
 created by a broker to handle*/
    Map<Endpoint, *? extends CompletionStage<Void>> start(AuthorizerServerInfo serverInfo);

    a/**
 client request to create ACLs.* ThisAuthorizes maythe bespecified nullaction. ifAdditional ACLsmetadata arefor createdthe directlyaction inis ZooKeeperspecified
     *        using AclCommandin `requestContext`.
     * @param<p>
 resourcePattern Resource pattern for which* ACLsThis areis beinga added
synchronous API designed for use *with @paramlocally accessControlEntriescached ListACLs. ofSince accessthis controlmethod entriesis toinvoked addon for resourcethe
     *
 request thread while processing * @return Create result for each access control entry in the same order as in the input listeach request, implementations of this method should avoid time-consuming
     * remote communication that may block request threads.
     */
    List<AclCreateResult> createAcls(KafkaRequestContext * @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 ResourcePatternresults resourcePattern,
for each action in the same order as the provided actions
     */
    List<AuthorizationResult> authorize(AuthorizableRequestContext requestContext,                List<AccessControlEntry> accessControlEntriesList<Action> actions);

    /**
     * DeletesCreates the specifiednew ACL bindingbindings.
     * <p>
     * @param requestContext Request context ifThis is an asynchronous API that enables the ACLcaller to isavoid beingblocking deletedduring bythe aupdate. brokerImplementations toof handlethis
     * API can return completed futures using  a client request to delete ACLs. This may be null if ACLs are deleted directly in ZooKeeper{@link java.util.concurrent.CompletableFuture#completedFuture(Object)}
     * to process the update synchronously on the request thread.
     *
     * @param requestContext usingRequest AclCommand.
context if the ACL is *being @paramcreated resourcePatternby Resourcea patternbroker forto whichhandle
 ACLs are being removed
 *    * @param accessControlEntries Set ofa accessclient controlrequest entries to removecreate forACLs. resource
This may be null if *
ACLs are created directly in *ZooKeeper
 @return Delete result for each access* control entry in the same order as in the input listusing AclCommand.
     * @param aclBindings ACL bindings to create
   Each result indicates*
 if the entry was actually* deleted.
@return Create result for each */
ACL binding in the List<AclDeleteResult>same deleteAcls(KafkaRequestContext requestContext,
      order as in the input list. Each result
     *         is returned as a CompletionStage that completes when the result is available.
      ResourcePattern resourcePattern,*/
    List<? extends CompletionStage<AclCreateResult>> createAcls(AuthorizableRequestContext requestContext,                             List<AccessControlEntry> accessControlEntriesList<AclBinding> aclBindings);

    /**
     * Deletes all ACL bindings forthat match the specifiedprovided filters.
 resource pattern. Only ACLs explicitly* configured<p>
     * withThis thisis resourcean pattern are deleted, no matching is performed.
     *asynchronous API that enables the caller to avoid blocking during the update. Implementations of this
     * @paramAPI requestContextcan Requestreturn contextcompleted iffutures theusing ACLs are being deleted by a broker to handle
     *  {@link java.util.concurrent.CompletableFuture#completedFuture(Object)}
     * to process the update synchronously on the request thread.
     *
 a client request to delete* ACLs.@param ThisrequestContext mayRequest becontext nullif the ifACL ACLsis arebeing deleted directly in ZooKeeperby a broker to handle
     *        usinga AclCommand.
client request to delete ACLs. *This @parammay resourcePatternbe Resourcenull patternif whose ACLs are deleted
 directly in ZooKeeper
  * @return true if* at least one ACL binding was deleted, falseusing otherwiseAclCommand.
     */
 @param aclBindingFilters Filters booleanto deleteAcls(KafkaRequestContext requestContext, ResourcePattern resourcePattern);

    /**match ACL bindings that are to be deleted
     * Returns all ACL bindings.
     *
 @return Delete result for *each @returnfilter Accessin controlthe entriessame oforder allas ACLin bindingsthe grouped by resource patterninput list.
     */
    Map<ResourcePattern, Set<AccessControlEntry>> acls();

   Each /**
result indicates which ACL bindings *were Returnsactually accessdeleted controlas entrieswell definedas forany
 the specified resource pattern.
 *    *
     *bindings @paramthat resourcePatternmatched Resourcebut patterncould fornot whichbe ACLsdeleted. areEach returned.result Nois matchingreturned isas performed,a
     *         onlyCompletionStage ACLsthat definedcompletes forwhen the specifiedresult patternis are returnedavailable.
     */
    List<? extends CompletionStage<AclDeleteResult>> deleteAcls(AuthorizableRequestContext *requestContext, @return ACL bindings for resource pattern.List<AclBindingFilter> aclBindingFilters);

    /**
     */
 Returns ACL bindings which Set<AccessControlEntry>match acls(ResourcePattern resourcePattern);
}

Request context will provided to authorizers using a new interface KafkaRequestContext. The existing class org.apache.kafka.common.requests.RequestContext will implement this interface.

package org.apache.kafka.common;

import java.net.InetAddress;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.apache.kafka.common.security.auth.SecurityProtocol;

/**
 * Request context interface that provides data from request header as well as connection
 * and authentication information to plugins.
 */
@InterfaceStability.Evolving
public interface KafkaRequestContext {

    /**
     * Returns name of listener on which request was received.the provided filter.
     * <p>
     * This is a synchronous API designed for use with locally cached ACLs. This method is invoked on the request
     * thread while processing DescribeAcls requests and should avoid time-consuming remote communication that may
     */
 block request threads.
 String listenerName();

    /**
     * Returns the security protocol@return Iterator for theACL listenerbindings, onwhich whichmay requestbe waspopulated receivedlazily.
     */
    SecurityProtocolIterable<AclBinding> securityProtocolacls(AclBindingFilter filter);
}

Request context will be provided to authorizers using a new interface AuthorizableRequestContext. The existing class org.apache.kafka.common.requests.RequestContext will implement this interface. New methods may be added to this interface in future, so mock implementations using this interface should adapt to these changes.

package org.apache.kafka.server.authorizer;

import java.net.InetAddress;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.apache.kafka.common.security.auth.SecurityProtocol;

/**
 * Request context interface that provides data from request header as well as connection
 * and authentication information to plugins.
 */
@InterfaceStability.Evolving
public interface AuthorizableRequestContext {

        /**
     * Returns authenticated principal for the connection on which request was received.
     */
    KafkaPrincipal principal();

    /**
     * Returns clientname IPof addresslistener fromon which request was sentreceived.
     */
    InetAddressString clientAddresslistenerName();

    /**
     * Returns the requestsecurity typeprotocol (ApiKey)for fromthe thelistener requeston header.which Forrequest fetchwas requests,received.
     */
 the metrics names FetchFollower and FetchConsumer will be used to distinguish betweenSecurityProtocol securityProtocol();

    /**
     * replicaReturns fetchauthenticated requestsprincipal and client fetch requestsfor the connection on which request was received.
     */
    StringKafkaPrincipal requestTypeprincipal();

    /**
     * Returns theclient requestIP versionaddress from thewhich request was headersent.
     */
    intInetAddress requestVersionclientAddress();

    /**
     * Returns16-bit API key of the clientrequest id from the request header. See
     * https://kafka.apache.org/protocol#protocol_api_keys for request types.
     */
    Stringint clientIdrequestType();

    /**
     * Returns the correlationrequest idversion from the request header.
     */
    int correlationIdrequestVersion();
}

Action  provides details of the action being authorized including resource and operation. Additional context including authorization mode are also included, enabling access violation to be distinguished from resource filtering or optional ACLs.

    /**
     * Returns the client id from the request header.
     */
    String clientId();

    /**
     * Returns the correlation id from the request header.
     */
    int correlationId();
}

AuthorizerServerInfo provides runtime broker configuration to authorization plugins including broker id, cluster id and endpoint information. New methods may be added to this interface in future, so mock implementations using this interface should adapt to these changes.

package org.apache.kafka.server.authorizer;

import java.util.Collection;
import org.apache.kafka.common.ClusterResource;
import org.apache.kafka.common.Endpoint;
import org.apache.kafka.common.annotation.InterfaceStability;

/**
 * Runtime broker configuration metadata provided to authorizers during start up.
 */
@InterfaceStability.Evolving
public interface AuthorizerServerInfo {

    /**
     * Returns cluster metadata for the broker running this authorizer including cluster id.
     */
    ClusterResource clusterResource();

    /**
     * Returns broker id. This may be a generated broker id if `broker.id` was not configured.
     */
    int brokerId();

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;
    }

    /**
     * Returns Resourceendpoints onfor whichall actionlisteners isincluding beingthe performed.
advertised host and port to */which
    public * ResourcePattern resourcePattern() {
the listener is bound.
     */
    returnCollection<Endpoint> resourcePattern;
    }endpoints();

    /**
     * Returns Operationthe beinginter-broker performedendpoint.
 This is one of */
the endpoints returned by public AclOperation operation{@link #endpoints() {}.
     */
    returnEndpoint operationinterBrokerEndpoint();
    }

    /**
     * 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:

}

Endpoint is added as a common class so that it may be reused in several places in the code where we use this abstraction.

package org.apache.kafka.common;

import java.util.Objects;
import java.util.Optional;

import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.security.auth.SecurityProtocol;

/**
 * Represents a broker endpoint.
 */

@InterfaceStability.Evolving
public class Endpoint {

    private final String listenerName;
    private final SecurityProtocol securityProtocol;
    private final String host;
    private final int port;

    public Endpoint(String listenerName, SecurityProtocol securityProtocol, String host, int port) {
        this.listenerName = listenerName;
        this.securityProtocol = securityProtocol;
        this.host = host;
        this.port = port;
    }

    /**
     * Returns the listener name of this endpoint. This is non-empty for endpoints provided
     * to broker plugins, but may be empty when used in clients.
     */
    public Optional<String> listenerName() {
        return Optional.ofNullable(listenerName);
    }

    /**
     * Returns the security protocol of this endpoint.
     */
    public SecurityProtocol securityProtocol() {
        return securityProtocol;
    }

    /**
     * Returns advertised host name of this endpoint.
     */
    public String host() {
        return host;
    }

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 DENIED, request is failed with authorization failure.
     */
    MANDATORY,

    /**
     * Access was requested to resource. If authorization result is ALLOWED, access is granted to
     * the resource to perform the request. If DENIED, alternative authorization rules are applied
     * to determine if access is allowed.
     */
    OPTIONAL,

    /**
     * AccessReturns wasthe requestedport to authorized resources (e.g. to subscribe to regex pattern).which the listener is bound.
     */
    public int * Request is performed on resources whose authorization result is ALLOWED and the rest of
     * the resources are filtered out.port() {
        return port;
    }

    @Override
    public boolean equals(Object o) {
     */
    FILTER,

    /**if (this == o) {
     * Request to list authorized operations. No accessreturn istrue;
 actually performed by this    request}
      *  based on the authorization result.
if (!(o instanceof Endpoint)) {
       */
     LIST_AUTHORIZED
}

...

return

...

package org.apache.kafka.server.authorizer;

public enum AuthorizationResult {
 false;
        ALLOWED,}

    DENIED
}

public class AclCreateResult {
 Endpoint that = private final Throwable exception(Endpoint) o;

    public AclCreateResult() {
  return Objects.equals(this.listenerName, that.listenerName) &&
   this(null);
    }

    public AclCreateResult(Throwable exception) {Objects.equals(this.securityProtocol, that.securityProtocol) &&
        this.exception = exception;
    }
Objects.equals(this.host, that.host) &&
    /**
     * Returns any exception during createthis.port If exception is null, the request has succeeded.== that.port;

    }

     */@Override
    public Throwableint exceptionhashCode() {
        return exceptionObjects.hash(listenerName, securityProtocol, host, port);
    }

    /**@Override
    public *String Returns true if the request failed.
toString() {
        return  */"Endpoint(" +
    public boolean failed() {
        return exception != null;
    }
}

public class AclDeleteResult {
"listenerName='" + listenerName + '\'' +
       private final Throwable exception;
  ", securityProtocol=" private+ finalsecurityProtocol boolean deleted;

+
     public AclDeleteResult(Throwable exception) {
    ", host='" + host this(false, exception);
    }

+ '\'' +
    public AclDeleteResult(boolean deleted) {
     ", port=" + this(deleted, null);port +
    }

    private AclDeleteResult(boolean deleted, Throwable exception) {
        this.deleted = deleted')';
        this.exception = exception;
    }

    /**
     * Returns any exception while attempting to delete an ACL.
     */
    public Throwable exception() {
        return exception;
    }

    /**
     * Returns true if request has failed with an exception. {@link #deleted()} indicates if
     * an ACL was deleted.
     */
    public boolean failed() {
}
}

Action  provides details of the action being authorized including resource and operation. Additional context including audit flag indicating authorization usage are also included, enabling access violation to be distinguished from resource filtering or optional ACLs.

package org.apache.kafka.server.authorizer;

import java.util.Objects;
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 int resourceReferenceCount;
    private final boolean logIfAllowed;
    private final boolean logIfDenied;

    public Action(AclOperation operation,
                  ResourcePattern resourcePattern,
                return exception != null;
int resourceReferenceCount,
           }

    /**
   boolean logIfAllowed,
 * Returns true if a delete was performed.
     */
    public boolean deleted(logIfDenied) {
        return deletedthis.operation = operation;
    }
}

Proposed Changes

Deprecate existing Scala Authorizer

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

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

kafka.server.KafkaServer will be updated to provide cluster-id to the configured authorizer if it implements org.apache.kafka.common.ClusterResourceListener interface. This will enable the authorizer to include cluster-id in logs and to use centralized ACLs if required.

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.

Compatibility, Deprecation, and Migration Plan

What impact (if any) will there be on existing users?

Existing authorizer interfaces and classes are being deprecated, but not removed. We will continue to support the old 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.

Test Plan

All the existing integration and system tests will be updated to use the new config and the new authorization class. Unit tests will be added for testing the new methods and parameters being introduced in this KIP. An additional integration test will be added to test authorizers using the old Scala API.  

Rejected Alternatives

Description of Authorizer as proposed in KIP-50

KIP-50 proposes to return a textual description of the authorizer that can be used in tools like AclCommand. Since we don’t support returning a description using AdminClient and none of the other pluggable APIs have similar support, this KIP does not add a method to return authorizer description.

Separate authorizers for each listener

    this.resourcePattern = resourcePattern;
        this.logIfAllowed = logIfAllowed;
        this.logIfDenied = logIfDenied;
        this.resourceReferenceCount = resourceReferenceCount;
    }

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

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

    /**
     * Indicates if audit logs tracking ALLOWED access should include this action if result is
     * ALLOWED. The flag is true if access to a resource is granted while processing the request as a
     * result of this authorization. The flag is false only for requests used to describe access where
     * no operation on the resource is actually performed based on the authorization result.
     */
    public boolean logIfAllowed() {
        return logIfAllowed;
    }

    /**
     * Indicates if audit logs tracking DENIED access should include this action if result is
     * DENIED. The flag is true if access to a resource was explicitly requested and request
     * is denied as a result of this authorization request. The flag is false if request was
     * filtering out authorized resources (e.g. to subscribe to regex pattern). The flag is also
     * false if this is an optional authorization where an alternative resource authorization is
     * applied if this fails (e.g. Cluster:Create which is subsequently overridden by Topic:Create).
     */
    public boolean logIfDenied() {
        return logIfDenied;
    }

    /**
     * Number of times the resource being authorized is referenced within the request. For example, a single
     * request may reference `n` topic partitions of the same topic. Brokers will authorize the topic once
     * with `resourceReferenceCount=n`. Authorizers may include the count in audit logs.
     */
    public int resourceReferenceCount() {
        return resourceReferenceCount;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (!(o instanceof Action)) {
            return false;
        }

        Action that = (Action) o;
        return Objects.equals(this.resourcePattern, that.resourcePattern) &&
            Objects.equals(this.operation, that.operation) &&
            this.resourceReferenceCount == that.resourceReferenceCount &&
            this.logIfAllowed == that.logIfAllowed &&
            this.logIfDenied == that.logIfDenied;

    }

    @Override
    public int hashCode() {
        return Objects.hash(resourcePattern, operation, resourceReferenceCount, logIfAllowed, logIfDenied);
    }

    @Override
    public String toString() {
        return "Action(" +
            ", resourcePattern='" + resourcePattern + '\'' +
            ", operation='" + operation + '\'' +
            ", resourceReferenceCount='" + resourceReferenceCount + '\'' +
            ", logIfAllowed='" + logIfAllowed + '\'' +
            ", logIfDenied='" + logIfDenied + '\'' +
            ')';
    }
}

Authorize method returns individual allowed/denied results for every action.

package org.apache.kafka.server.authorizer;

public enum AuthorizationResult {
    ALLOWED,
    DENIED
}

ACL create operation returns any exception from each ACL binding requested.

package org.apache.kafka.server.authorizer;

import java.util.Optional;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.errors.ApiException;

@InterfaceStability.Evolving
public class AclCreateResult {
    public static final AclCreateResult SUCCESS = new AclCreateResult();

    private final ApiException exception;

    private AclCreateResult() {
        this(null);
    }

    public AclCreateResult(ApiException exception) {
        this.exception = exception;
    }

    /**
     * Returns any exception during create. If exception is empty, the request has succeeded.
     */
    public Optional<ApiException> exception() {
        return exception == null ? Optional.empty() : Optional.of(exception);
    }
}

ACL delete operation returns any exception from each ACL filter requested. Matching ACL bindings for each filter are returned along with any delete failure.

package org.apache.kafka.server.authorizer;

import java.util.Collections;
import java.util.Collection;
import java.util.Optional;
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 Optional<ApiException> exception() {
        return exception == null ? Optional.empty() : Optional.of(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 any exception that resulted in failure to delete ACL binding.
         */
        public Optional<ApiException> exception() {
            return exception == null ? Optional.empty() : Optional.of(exception);
        }
    }
}

Proposed Changes

Asynchronous update requests

kafka.server.KafkaApis will be updated to handle CreateAcls and DeleteAcls requests asynchronously using a purgatory. If Authorizer.createAcls or Authorizer.deleteAcls returns any ComplettionStage that is not complete, the request will be added to a purgatory and completed when all the stages complete. Authorizer implementations with low latency updates may continue to update synchronously and return a completed future. These requests will be completed in-line and will not be added to the purgatory.

Asynchronous updates are useful for Authorizer implementations that use external stores for ACLs, for example a database. Async handling of update requests will enable Kafka brokers to handle database outages without blocking request threads. As many databases now support async APIs (https://dev.mysql.com/doc/x-devapi-userguide/en/synchronous-vs-asynchronous-execution.html, https://blogs.oracle.com/java/jdbc-next:-a-new-asynchronous-api-for-connecting-to-a-database), async update API enables authorizers to take advantage of these APIs.

Purgatory metrics will be added for ACL updates, consistent with metrics from other purgatories. Two new metrics will be added:

• kafka.server:type=DelayedOperationPurgatory,name=NumDelayedOperations,delayedOperation=acl-update
• kafka.server:type=DelayedOperationPurgatory,name=PurgatorySize,delayedOperation=acl-update

In addition to these metrics, existing request metrics for CreateAcls and DeleteAcls can be used to track the portion of time spent on async operations since local time is updated before the async wait and remote time is updated when async wait completes:

• kafka.network:type=RequestMetrics,name=LocalTimeMs,request=CreateAcls
• kafka.network:type=RequestMetrics,name=RemoteTimeMs,request=CreateAcls
• kafka.network:type=RequestMetrics,name=LocalTimeMs,request=DeleteAcls
• kafka.network:type=RequestMetrics,name=RemoteTimeMs,request=DeleteAcls

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.

AclAuthorizer

A new authorizer implementation will be added. kafka.security.authorizer.AclAuthorizer will implement the new interface, making use of the additional request context available to improve authorization logging. This authorizer will be 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

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.

Compatibility, Deprecation, and Migration Plan

What impact (if any) will there be on existing users?

Existing authorizer interfaces and classes are being deprecated, but not removed. We will continue to support the 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 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.

Test Plan

All the existing integration and system tests will be updated to use the new config and the new authorization class. Unit tests will be added for testing the new methods and parameters being introduced in this KIP. An additional integration test will be added to test authorizers using the old Scala API.  

Rejected Alternatives

Description of Authorizer as proposed in KIP-50

KIP-50 proposes to return a textual description of the authorizer that can be used in tools like AclCommand. Since we don’t support returning a description using AdminClient and none of the other pluggable APIs have similar support, this KIP does not add a method to return authorizer description.

Separate authorizers for each listener

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.

Make authorize() asynchronous

Authorize operations in the existing Authorizer are synchronous and this KIP proposes to continue to authorize synchronously on the request thread while processing each request. This requires all ACLs to be cached in every broker to avoid blocking request threads during authorization. To improve scalability in future, we may want to support asynchronous authorize operations that may perform remote communication, for example with an LRU cache. But asynchronous authorize operations add complexity to the Kafka implementation. Even though we may be able to use the existing purgatory, additional design work is required to figure out how this can be implemented efficiently.  So it was decided that we should keep the authorization API synchronous for now. In future, we can add async authorize as a new method on the API if requiredIn 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.