...

This support would greatly simplify ACL operational story in a multi-tenant environment.

Public Interfaces

• We would will add a new methods method getMatchingAcls(resource) to the Authorizer interface.

  Code Block
  language scala title Authorizer
  /** * Get set of all acls that match this resource. * @param resource regular or wildcard-suffixed resource name. * @return empty set if no acls are found, otherwise the acls for the matching resources. */ def getMatchingAcls(resource: Resource): Set[Acl]

• Changes to command line tool class https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/admin/AclCommand.scala to expose the above new APIsAPI.
• Changes to AdminClient (needs discussion).
• (Question) Should we deprecate older getAcls(resource) method? Not at this point?
• Update the public documentation with the details of the new feature.

Proposed Changes

Solution

The proposal is to extend the concept of wildcard ACL (‘*’) to support wildcard-suffixed ACLs (‘name*’).
This means that it would be possible to create ACLs of type: User:clientA has READ access on topic orgA* from hostA,
i.e clientA has READ access to all topics that start with `orgA` from hostA.
The concept of wildcard-suffixed ACLs would be applicable only to ONLY resource resource names.

Storage model

Currently, ACLs are stored on ZK under path /kafka-acl/<resource-type>/<resource-name>.

For example:
ACLs for topic topicName would be stored under /kafka-acl/Topic/topicName.
ACLs for consumer group groupId would be stored under /kafka-acl/Group/groupId.

An example ACL definition looks like:

$ get /kafka-acl/Topic/topicName
{"version":1,"acls":[{"principal":"User:clientA","permissionType":"Allow","operation":"Read","host":"*"},{"principal":"User:clientA","permissionType":"Allow","operation":"Write","host":"*"},{"principal":"clientB","permissionType":"Allow","operation":"Write","host":"host1"}]}

Current supported resource names are either full resource names like topicName or a special wildcard '*'.

$ get /kafka-acl/Topic/*
{"version":1,"acls":[{"principal":"User:clientA","permissionType":"Allow","operation":"Read","host":"*"}]}
which means that clientA has read access to all topics from all hosts.

Now, we extend the same storage model to store ACLs under wildcard-suffixed resource names like orgName*.

$ get /kafka-acl/Topic/teamA*
{"version":1,"acls":[{"principal":"User:clientA","permissionType":"Allow","operation":"Read","host":"*"}]}

ACLs write path

The above storage model ensures that there are NO changes on the ACL write path.

Input resourceName would be treated the same way as earlier i.e. the appropriate ACLs would be rewritten at /kafka-acl/<resource-type>/<resource-name> location.

$ get /kafka-acl/Topic/orgName*
{"version":1,"acls":[{"principal":"User:clientA","permissionType":"Allow","operation":"Read","host":"*"}]}

ACLs read path

On read path, we look for all matching ACLs when:

a) getMatchingAcls(resourceWithWildcardSuffix) is called.
b) authorize(…) is called.

Access would be allowed if there is at least one ALLOW matching acl and no DENY matching acl (current behavior is maintained). Note that the length of the prefix doesn't play any role here.

...