Kafka支持基于SSL和基于SASL的安全认证机制。

基于SSL的认证主要是指Broker和客户端的双路认证。即客户端认证broker的证书，broker也认证客户端的证书。

Kafka还支持通过SASL做客户端认证。SASL是提供认证和数据安全服务的框架。Kafka支持的SASL机制有5种：
1.GSSAPI：也就是 Kerberos 使用的安全接口。
2.PLAIN：是使用简单的 用户名/密码 认证的机制。（不能动态的增减kafka用户，必须重启 Kafka 集群才能令变更生效，因为认证用户信息全部保存在静态文件中）
3.SCRAM：主要用于解决PLAIN机制安全问题的新机制。
4.OAUTHBEARER：是基于 OAuth2 认证框架的新机制。
5.Delegation Token：补充现有SASL机制的轻量级认证机制。

KAFKA认证步骤：

SASL/SCRAM-SHA-256 配置实例

创建用户：

./kafka-configs.sh --bootstrap-server 192.168.1.10:9192 --alter --add-config 'SCRAM-SHA-256=[password=admin],SCRAM-SHA-512=[password=admin]' --entity-type users --entity-name admin

./kafka-configs.sh --bootstrap-server 192.168.1.10:9192 --alter --add-config 'SCRAM-SHA-256=[password=reader],SCRAM-SHA-512=[password=reader]' --entity-type users --entity-name reader

./kafka-configs.sh --bootstrap-server 192.168.1.10:9192 --alter --add-config 'SCRAM-SHA-256=[password=write],SCRAM-SHA-512=[password=write]' --entity-type users --entity-name write

查看创建的用户：

bin/kafka-configs.sh --zookeeper 192.168.1.10:2181 --describe --entity-type users  --entity-name write

创建JAAS文件：（为每个broker都需要创建）

KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="admin";
};
KafkaClient {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="admin";
};

修改server.properties 文件

sasl.enabled.mechanisms=SCRAM-SHA-256
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
security.inter.broker.protocol=SASL_PLAINTEXT
listeners=SASL_PLAINTEXT://192.168.1.10:9092

启动kafka：
修改kafka-run-class.sh （大致219行）

if [ -z "$KAFKA_OPTS" ]; then
  KAFKA_OPTS="-Djava.security.auth.login.config=/zb/kafka/kafka_2.12-2.7.2-test1/broker.jaas"
fi

启动：
nohup bin/kafka-server-start.sh config/server.properties &

测试：

发送消息：
创建一个producer.conf配置文件：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="writer" password="writer";

启动生产者：

bin/kafka-console-producer.sh --broker-list 192.168.1.10:9192,192.168.1.10:9292,192.168.1.10:9392 --topic kl-test1 --producer.config /zb/kafka/kafka_2.12-2.7.2-test1/producer.conf

消费消息：
创建一个consumer.conf 的脚本

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="reader" password="reader";

启动消费者：

bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.10:9192,192.168.1.10:9292,192.168.1.10:9392 --topic kl-test1 --from-beginning --consumer.config /zb/kafka/kafka_2.12-2.7.2-test1/consumer.conf 

查看topic，创建管理者认证：
创建一个admin配置文件：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256

./bin/kafka-topics.sh --bootstrap-server 192.168.1.10:9192 --command-config /zb/kafka/kafka_2.12-2.7.2-test1/admin_sals.conf --list

动态增减用户：

删除用户

bin/kafka-configs.sh --zookeeper 192.168.1.10:2181 --alter --delete-config 'SCRAM-SHA-256' --entity-type users --entity-name writer
bin/kafka-configs.sh --zookeeper 192.168.1.10:2181 --alter --delete-config 'SCRAM-SHA-512' --entity-type users --entity-name writer

添加新用户 

bin/kafka-configs.sh --bootstrap-server 192.168.1.10:9192 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=new_writer]' --entity-type users --entity-name new_writer