linux kafka 创建用户

创建管理员

bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[password=admin-secret],SCRAM-SHA-512=[password=admin-secret]' --entity-type users --entity-name admin

#创建客户端用户fanboshi

bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=fanboshi],SCRAM-SHA-512=[password=fanboshi]' --entity-type users --entity-name fanboshi

#查看证书

bin/kafka-configs.sh --zookeeper localhost:2182 --describe --entity-type users --entity-name fanboshi

#配置JAAS文件

[root@node002229 config]# cat kafka_server_jaas.conf
KafkaServer {
  org.apache.kafka.common.security.scram.ScramLoginModule required
  username="admin"
  password="admin-secret";

配置SASL端口和SASL机制

vim /usr/local/kafka/bin/kafka-server-start.sh
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=$base_dir/../config/kafka_server_jaas.conf kafka.Kafka "$@"

[root@node002229 config] vim server.properties
listeners=SASL_PLAINTEXT://0.0.0.0:9092
advertised.listeners=SASL_PLAINTEXT://127.0.0.1:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
sasl.enabled.mechanisms=SCRAM-SHA-256
super.users=User:admin
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=false

重启zookeeper 和 kafka

sh bin/zkServer.sh restart
sh bin/kafka-server-stop.sh
sh bin/kafka-server-start.sh config/server.properties

创建kafka_client_jaas_admin.conf文件

[root@node002229 kafka]# vim config/kafka_client_jaas_fanboshi.conf 
KafkaClient {
  org.apache.kafka.common.security.scram.ScramLoginModule required
  username="fanboshi"
  password="fanboshi";
};

创建kafka-console-producer-fanboshi.sh

[root@node002229 bin]vim kafka-console-producer-fanboshi.sh
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx512M"
fi
exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=$(dirname $0)/../config/kafka_client_jaas_fanboshi.conf kafka.tools.ConsoleProducer "$@"

创建 kafka-console-consumer-fanboshi.sh 文件

[root@node002229 bin]vim kafka-console-consumer-fanboshi.sh
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx512M"
fi
exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=$(dirname $0)/../config/kafka_client_jaas_fanboshi.conf kafka.tools.ConsoleConsumer"$@"

创建topic

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

添加生产者权限

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=127.0.0.1:2181 --add --allow-principal User:fanboshi --producer --topic "test"

添加消费者权限

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=127.0.0.1:2181 --add --allow-principal User:fanboshi --consumer --topic "test" --group '*'

添加事务权限

./kafka-acls.sh 
--authorizer-properties zookeeper.connect=localhost:2181 
--add 
--allow-principal User:fanboshi 
--topic test 
--producer 
--transactional-id '*'