...

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

Public Interfaces

Summary of changes

• Add new field 'ResourceNameType' to Resource, and ResourceFilter classes to distinguish between literal and prefixed resource names.
  • ResourceNameType is an enum to support more types in the future.
  • ResourceNameType will include an 'ANY' type for use in filters
  • ResourceNameType will default to literal, to maintain backwards compatibility with existing clients.

...

• Changes to command line tool class https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/admin/AclCommand.scala
  • Expose a '--resource-name-type' flag, which can be set to literal, prefixed or any, e.g.
          bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:Bob  --operation Read --group my-app-* --resource-name-type prefixed
  • The flag will default to 'literal', meaning the commands will return only the ACLs they previously returned, even if prefixed ACLs exists for the resource.
  • Users can now set the flag to 'all' to retrieve/delete all ACLs affecting the supplied ACLs resource.
    
    
• Changes to the admin client to support prefixed ACLs.
  • New schema version for CreateAclsRequest / DeleteAclsRequest / DescribeAclsRequest, and associated responses, which will have a new byte field in schemas to distinguish literals vs prefixed resource names.
  • The CreateAclRequest will only accept prefixed or literal ResourceNameTypes
  • The DeleteAclRequest and DescribeAclRequests will accept prefixed or literal, and will return delete/describe only ACLs exactly matching the resources. i.e. no pattern matching will be performed
  • The DeleteAclRequest and DescribeAclRequests will also accept a ResourceNameType of 'all', which will delete/describe all ACLs affecting the resource, including the literal wildcard ACL '*', if present.

AdminClient changes

The examples below should help demonstrate the purposed functionality:

Exissting functionality

Each call retrieves ACLs stored in one path in ZK:
adminClient.describeAcls(... TOPIC, "foobar" ...) -> would return only ACLs from '/kafka-acls/foobar' path
adminClient.describeAcls(... TOPIC, "*" ...) -> would return only ACLs from '/kafka-acls/*' path

New functionality

Using legacy constructors

Legacy constructors default resourceNameType to Literal and maintains existing contract:
adminClient.describeAcls(... TOPIC, "foobar" ...) -> still returns only ACLs from '/kafka-acls/foobar' path
adminClient.describeAcls(... TOPIC, "*" ...) -> still returns only ACLs from '/kafka-acls/*' path

Using constructors with explicit resourceNameType

The user needs to know, up front, which prefixed resource paths exist to be able to query them:
adminClient.describeAcls(... TOPIC, "foobar", Literal ...)-> will return only ACLs from '/kafka-acls/foobar' path
adminClient.describeAcls(... TOPIC, "*", Literal ...) -> will return only ACLs from '/kafka-acls/*' path
adminClient.describeAcls(... TOPIC, "foo", Prefixed ...) -> will return only ACLs from '/kafka-prefixed-acls/foo' path

Using 'ANY' resource name type

This allows user to discover all the ACLs affecting a specific resource:
adminClient.describeAcls(... TOPIC, "foobar", Any ...) -> will return all ACLs affecting topic "foobar", including any prefixed and wildcard ACLs.

The return value from describeAcls will contain the resourceNameType field for each ACL, so the user can determine if it is literal or prefixed.

Acls on prefixed resource paths are never returned to earlier clients. Nor can older clients delete ACLs on prefixed resource paths.

Proposed Changes

Solution

The proposal is to enhance the SimpleAclAuthorizer to support prefixed ACLs.
This means that it will be possible to create ACLs of type: User:clientA has READ access on topic prefixed with 'orgA' from 'hostA', i.e clientA has READ access to all topics that start with `orgA` from hostA.
The concept of prefixed ACLs will be applicable only to resource names.

Storage model

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

For example:
ACLs for topic topicName will be stored under /kafka-acl/Topic/topicName.
ACLs for consumer group groupId will 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.

...