kafka ACL常用权限操作
- 创建topic
使用bin/kafka-topics.sh创建
注意工具bin/kafka-topics.sh访问的是zookeeper而不是kafka，即他是一个zookeeper client而不是一个kafka client，所以它的认证都是通过zookeeper完成的。
Case 1：如果zookeeper没有配置ACL激活：
/opt/kafka/bin/kafka-topics.sh --create \
--zookeeper zookeeper.example.com \
--replication-factor 1 \
--partitions 1 \
--topic kafkaclient-topic
Case 2：如果zookeeper已经配置ACL激活：
命令还是前面的那个命令，但是必须提供java.security.auth.login.config指向jaas.conf文件。例如：
$ cat /path/to/zookeeper_client_jaas_admin.conf
Client {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="adminpwd";
};
命令的配置可以直接修改jvm的启动脚本，或者设置在环境变量里：
export KAFKA_OPTS=-Djava.security.auth.login.config=/path/to/zookeeper_client_jaas_admin.conf
这里配置的用户必须是zookeeper服务端已经配置运行可以访问的客户端用户。例如，下面的zookeeper服务端配置：
$ cat /path/to/zookeeper_jaas.conf
Server {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="adminpwd"
user_admin="adminpwd";
};
运行客户端为admin的用户作为zookeeper客户端连接访问。
- 查询topic
$ /opt/kafka/bin/kafka-topics.sh --list --zookeeper zookeeper.example.com:2181
查询topic操作的ACL认证，同前面创建topic操作的认证一样，不细说，参考前面。
- 删除topic
/opt/kafka/bin/kafka-topics.sh \
--delete \
--zookeeper zookeeper.example.com:2181 \
--topic kafkaclient-topic
删除topic操作的ACL认证，同前面创建topic操作的认证一样，不细说，参考前面。
- 生产者topic
producer用的脚本是/opt/kafka/bin/kafka-console-producer.sh，注意这个producer脚本是和kafka打交道的(相对bin/kafka-topics.sh是和zookeeper打交道的)，所以：
- kafka-topics.sh 是作为zookeeper client，需要验证zookeeper的ACL信息。
- kafka-console-producer.sh 是作为kafka client，需要验证kafka的ACL信息。
命令行格式：
/opt/kafka/bin/kafka-console-producer.sh \
--broker-list kafka.example.com:9092 \
--topic kafkaclient--topic \
--producer.config /path/to/client-sasl.properties
文件/path/to/client-sasl.properties
$ cat /path/to/client-sasl.properties
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
还需要配置client用户信息，并传给JVM参数：
$ cat /path/to/kafka_client_jaas_kafkaclient.conf
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="kafkaclient"
password="kafkaclientpwd";
};
$ export KAFKA_OPTS="-Djava.security.auth.login.config=/path/to/kafka_client_jaas_kafkaclient.conf"
此时如果没有授权，则会得到如下错误信息：
[2018-12-27 04:35:58,877] WARN [Producer clientId=console-producer]
Error while fetching metadata with correlation id 1 :
{kafkaclient--topic=TOPIC_AUTHORIZATION_FAILED}
(org.apache.kafka.clients.NetworkClient)
[2018-12-27 04:35:58,883] ERROR Error when sending message to topic kafkaclient--topic
with key: null, value: 3 bytes with error:
(org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.errors.TopicAuthorizationException:
Not authorized to access topics: [kafkaclient--topic]
赋予producer的权限：
/opt/kafka/bin/kafka-acls.sh \
--authorizer-properties zookeeper.connect=zookeeper.example.com:2181
--add --allow-principal User:"kafkaclient" \
--producer \
--topic "kafkaclient--topic"
这个选项--producer实际上在Topic域上创建了(Write/Describe/Create)3个子权限：
Current ACLs for resource `Topic:LITERAL:kafkaclient--topic`:
User:kafkaclient has Allow permission for operations: Write from hosts: *
User:kafkaclient has Allow permission for operations: Describe from hosts: *
User:kafkaclient has Allow permission for operations: Create from hosts: *
当然用户也可以单独创建者三个子权限。
- 消费者topic
consumer用的脚本是/opt/kafka/bin/kafka-console-consumer.sh，注意和生产者producer一样，consumer也是和kafka打交道的(相对于bin/kafka-topics.sh是和zookeeper打交道的)，所以：
- kafka-topics.sh 是作为zookeeper client，需要验证zookeeper的ACL信息。
- kafka-console-producer.sh 是作为kafka client，需要验证kafka的ACL信息。
- kafka-console-consumer.sh是作为kafka client，需要验证kafka的ACL信息。
命令行格式：
/opt/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server kafka.example.com:9092 \
--topic kafkaclient--topic \
--consumer.config sasl/client-sasl.properties \
--from-beginning
选项--from-begining可以调整成其他值；配置文件/path/to/client-sasl.properties和producer的一样，不细说，参考生产者。
此时如果没有授权，则会得到如下错误信息：
[2018-12-27 05:59:37,481] WARN [Consumer clientId=consumer-1, groupId=console-consumer-28240]
Error while fetching metadata with correlation id 2 :
{kafkaclient--topic=TOPIC_AUTHORIZATION_FAILED}
(org.apache.kafka.clients.NetworkClient)
[2018-12-27 05:59:37,485] ERROR Error processing message, terminating consumer process:
(kafka.tools.ConsoleConsumer$)
org.apache.kafka.common.errors.TopicAuthorizationException:
Not authorized to access topics: [kafkaclient--topic]
赋予consumer的权限：
/opt/kafka/bin/kafka-acls.sh \
--authorizer-properties zookeeper.connect=zookeeper.example.com:2181 \
--add --allow-principal User:"kafkaclient" \
--consumer \
--topic "kafkaclient--topic" \
--group '*'
和producer相比，consumer还有一个额外的参数--group，如果没有限制，则置成'*'即可；这个--consumer的选择实际上在Topic域上创建了(Read/Describe)2个子权限，然后在Group域创建了(Read)1个子权限：
Current ACLs for resource `Topic:LITERAL:kafkaclient--topic`:
User:kafkaclient has Allow permission for operations: Describe from hosts: *
User:kafkaclient has Allow permission for operations: Read from hosts: *
Current ACLs for resource `Group:LITERAL:*`:
User:kafkaclient has Allow permission for operations: Read from hosts: *
这个地方我们注意一下，consumer没有Create的权限，所以如果kafka配置成auto.create.topics.enable=true，而此时topic不存在，那么consumer试图创建topic的时候会失败，那就需要一条单独的Create授权规则来给consumer增加Create权限。
- 权限管理工具 /opt/kafka/bin/kafka-acls.sh
权限管理工具以命令行的方式管理权限，可以增加/删除/列举所有的权限规则。
基本用法：
- 增加权限
/opt/kafka/bin/kafka-acls.sh \
--authorizer-properties zookeeper.connect=zookeeper.example.com:2181 \
--add \
--allow-principal User:"kafkaclient" \
--operation Read \
--topic "kafkaclient--topic"
授权用户kafaclient具有Read topic kafaclient--topic的权限。
- 删除权限
/opt/kafka/bin/kafka-acls.sh \
--authorizer-properties zookeeper.connect=zookeeper.example.com:2181 \
--remove \
--allow-principal User:"kafkaclient" \
--operation Describe \
--topic kafkaclient--topic \
--force
删除用户kafaclient具有Describe topic kafaclient--topic的权限。
- 查询权限
/opt/kafka/bin/kafka-acls.sh \
--authorizer-properties zookeeper.connect=zookeeper.example.com:2181 \
--list \
--topic 'kafkaclient--topic'
查看当前在topic kafkaclient--topic上面的权限列表。
另外注意，和kafka-topics.sh一样，kafka-acls.sh也是直接访问zookeeper的，而不是访问kafka，所以它的认证方式和kafka-topics.sh是一样的：
- kafka-topics.sh 是作为zookeeper client，需要验证zookeeper的ACL信息。
- kafka-acls.sh 是作为zookeeper client，需要验证zookeeper的ACL信息。
- kafka-console-producer.sh 是作为kafka client，需要验证kafka的ACL信息。
- kafka-console-consumer.sh是作为kafka client，需要验证kafka的ACL信息。
详细的用法配置请参考kafka-topics.sh部分，不细说。
- ACL的--resource-pattern-type选项
在kafka2.0之后引入了--resource-pattern-type这个参数，可以针对特定的资源(topic)命名规则，例如前缀，来为某一类的topic添加规则。而之前的办法只能读完整的topic设置规则，字符''表示所有的，这不是规则表达式匹配任意字符的意思，而就是文本字符''。
例如：
/opt/kafka/bin/kafka-acls.sh \
--authorizer-properties zookeeper.connect=zookeeper.example.com:2181 \
--add \
--allow-principal User:"kafkaclient" \
--operation All \
--resource-pattern-type prefixed \
--topic "kafkaclient--" \
授权用户kafkaclient具有访问所有以'kafkaclient--'开头的topic的权限；这样带来的好处是，以后我们使得kafkaclient创建的topic全部以'kafkaclient--'开头，那么就不需要再为这些topic创建rule，一条rule就能够动态的管理新加的topic。
其实我对这个还是不满意，如果能够定义灵活规则就好了；因为上面的限制，我还是需要为每一个用户添加一条规则，而我想为所有的用户只用一条规则，这条规则就是：任何用户具有访问以这个用户名开头的所有的topic；这样不管以后新加topic还是新加用户，都不用再新加rule了。遗憾的是目前kafka还是不支持这个功能。类似：
--allow-principal User:"kafkaclient" \
--resource-pattern-type prefixed \
--topic "{USER}--" \
