Verify kafka-acls.sh is available:
# Check PATH
which kafka-acls.sh
# Or set KAFKA_HOME
export KAFKA_HOME=/path/to/kafkakfcli acl listkfcli acl list --resource-type topic
kfcli acl list --resource-type group
kfcli acl list --resource-type clusterkfcli acl list --name my-topickfcli acl list --principal User:alicekfcli acl list --resource-type topic --name orders --principal User:alicekfcli acl create \
--resource-type topic \
--resource-name orders \
--principal User:consumer-app \
--operation read \
--permission allowkfcli acl create \
--resource-type topic \
--resource-name orders \
--principal User:producer-app \
--operation write \
--permission allowkfcli acl create \
--resource-type group \
--resource-name my-consumer-group \
--principal User:consumer-app \
--operation read \
--permission allowkfcli acl create \
--resource-type topic \
--resource-name orders \
--principal User:admin \
--operation all \
--permission allowkfcli acl create \
--resource-type topic \
--resource-name orders \
--principal User:producer-app \
--operation write \
--permission allow \
--host 192.168.1.100kfcli acl create \
--resource-type topic \
--resource-name order \
--resource-pattern prefixed \
--principal User:consumer-app \
--operation read \
--permission allowkfcli acl create \
--resource-type topic \
--resource-name sensitive-topic \
--principal User:untrusted \
--operation read \
--permission denykfcli acl delete \
--resource-type topic \
--resource-name orders \
--principal User:consumer-app \
--operation read \
--permission allowkfcli acl delete \
--resource-type topic \
--resource-name orders \
--principal User:producer-app \
--operation write \
--permission allow \
--host 192.168.1.100topic- Topic resourcesgroup- Consumer group resourcescluster- Cluster-level resourcestransactionalid- Transactional ID resourcesdelegationtoken- Delegation token resources
read- Read from resourcewrite- Write to resourcecreate- Create resourcesdelete- Delete resourcesalter- Alter resource configurationdescribe- Describe resourcesclusteraction- Cluster-level operationsdescribeconfigs- Describe configurationsalterconfigs- Alter configurationsidempotentwrite- Idempotent writesall- All operations
allow- Grant accessdeny- Deny access
literal- Exact match (default)prefixed- Prefix matchmatch- Any match
# Allow write to topic
kfcli acl create --resource-type topic --resource-name orders \
--principal User:producer-app --operation write --permission allow
# Allow describe topic
kfcli acl create --resource-type topic --resource-name orders \
--principal User:producer-app --operation describe --permission allow# Allow read from topic
kfcli acl create --resource-type topic --resource-name orders \
--principal User:consumer-app --operation read --permission allow
# Allow consumer group operations
kfcli acl create --resource-type group --resource-name orders-consumers \
--principal User:consumer-app --operation read --permission allow# Grant all operations on all topics (use prefixed pattern with empty name)
kfcli acl create --resource-type topic --resource-name "*" \
--principal User:admin --operation all --permission allow \
--resource-pattern literal# Set KAFKA_HOME
export KAFKA_HOME=/path/to/kafka
# Or add to PATH
export PATH=$PATH:/path/to/kafka/binCheck broker configuration has:
authorizer.class.name=kafka.security.authorizer.AclAuthorizerEnsure you're running as a super user or have ACL management permissions.
Check error message for valid values and correct your command.
Verify ACL was created:
kfcli acl list --resource-type topic --name ordersTest with actual producer/consumer:
# As the user, try to produce/consume
kafka-console-producer --bootstrap-server localhost:9092 --topic ordersFor detailed documentation, see:
ACL_MANAGEMENT.md- Full guideP2.5_ACL_IMPLEMENTATION_SUMMARY.md- Implementation detailskfcli acl --help- Built-in help