根据需求，给 Kafka 集群配置不使用 Kerberos 的 SASL+ACL 。Kafka 配置部分参考资料比较充足，这里参考了胡夕老师的《Apache Kafka 实战》中相关章节，配置过程顺利。Kafka Manager监控配置了 SASL Kafka 集群部分的资料比较少，在认证部分出现较多问题，最终配置出一个“能用”版本。Kafka_2.11-1.0.0  kafka-manager-1.3.3.23。

1. Kafka SASL+ACL

通过 producer、consumer、admin 实现了写入、读取和 consumer-group 控制权限的分离。没有配置 zookeeper 的 SASL

Broker 端

1. 创建 JAAS 文件，注意结尾两个分号，保存为 kafka_server_jaas.conf

KafkaServer {
        org.apache.kafka.common.security.plain.PlainLoginModule required
        username="admin"
        password="admin"
        user_admin="admin"
        user_producer="prod-sec"
        user_consumer="cons-sec";
};

由于 kafka-server-start.sh 只接收 server.properties 的位置，故需要修改启动脚本，cp 一份修改最后一行为

exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=$KAFKA_HOME/config/kafka_server_jaas.conf kafka.Kafka "$@"

2. 修改 broker 启动所需的 server.properties 文件，前五条必须

authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
listeners=SASL_PLAINTEXT://ip:9092
security.inter.broker.protocol=SASL_PLAINTEXT 
sasl.enabled.mechanisms=PLAIN 
sasl.mechanism.inter.broker.protocol=PLAIN 
#设置超级用户
super.users=User:admin
#一个参数allow.everyone.if.no.acl.found
#设置为true，ACL机制改为黑名单机制，只有黑名单中的用户无法访问
#设置为false，ACL机制改为白名单机制，只有白名单中的用户可以访问，默认值为false

3. 使用新脚本启动 broker

启动zookeeper
kafka-server-start-sasl.sh config/server.properties

console-producer 脚本

1. 创建 kafka-prod-jaas.conf 认证文件

KafkaClient {
    org.apache.kafka.common.security.plain.PlainLoginModele required
    username="producer"
    password="ur password"
};

2. cp 一份脚本，修改最后一行

exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=$KAFKA_HOME/config/kafka_prod_jaas.conf kafka.tools.ConsoleProducer "$@"

3. 使用 ACL 配置 producer 写入授权

kafka-acls.sh --authorizer Kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:producer --operation Write --topic *

4. 使用修改后的 console-producer 脚本发送消息

kafka-console-producer-sasl.sh --broker-list localhost:9092 --topic test --producer-property security.protocol=SASL_PLAINTEXT --producer-property sasl.mechanism=PLAIN

 

console-cosnumer 脚本

1. 创建 kafka-cons-jaas.conf 认证文件

KafkaClient {
    org.apache.kafka.common.security.plain.PlainLoginModele required
    username="consumer"
    password="ur password"
};

2. cp 一份脚本，修改最后一行

exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=$KAFKA_HOME/config/kafka_cons_jaas.conf kafka.tools.ConsoleConsumer "$@"

3. 创建 consumer.config 文件或修改 /config/consumer.properties 文件

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
#group.id=test-group
#group可以指定，也可以不指定。
#一个问题：指定group后--from-beginning不生效，consumer会从记录的offset处继续消费

4. 使用 ACL 配置 consumer 读取授权

kafka-acls.sh --authorizer Kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:consumer --operation Read --topic * --group *

5. 使用修改后的 console-consumer 脚本接收消息

kafka-console-consumer-sasl.sh --bootstrap-server localhost:9092 --topic test --from-beginning --consumer.config consumer.properties

consumer-groups 脚本

1. 创建 admin_jaas.conf 认证文件

KafkaClient {
	org.apache.kafka.common.security.plain.PlainLoginModule required
	username="admin"
	password="ur password";
};

2. cp 一份脚本，修改最后一行

exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=$KAFKA_HOME/config/admin_jaas.conf kafka.admin.ConsumerGroupCommand "$@"

3. 创建 admin_sasl.config 文件

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

4. 使用修改后的 kafka-consumer-groups-sasl.sh 脚本 

kafka-consumer-groups-sasl.sh --bootstrap-server localhost:9092 --group test-group --describe --command-config $KAFAK_HOME/config/admin_sasl.config 

总结

SASL+ACL 是认证+授权的过程，除超级用户外的用户在认证后还应授予相应的权限

2. kafka-manager

在 Kafka 配置 SASL 后，需要对 kafka-manager 增加相应配置，监控 offset 等指标

无 SASL zookeeper ，在勾选 poll consumer info 后注意以下三个选项对应 broker 的配置即可