二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

i4T

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权 | i4T

https://i4t.com/users/auuAwtzuk · 2025-08-23 · via i4T

关于Kafka开启ACL权限控制，网上很多博客内容已经不适用于目前的版本，我这里使用官方最新版本3.9.0进行测试，并且3.1.0也测试没有问题，可放心使用

Kafka 开启SASL

在开启Kafka ACL之前，需要使用先开启SASL，否则无法直接开启ACL

详细SASL配置策略这里不在描述，可以查看下面的文章

Kafka 安全认证SASL

23年9月4日
喜欢：0
浏览：624

在启动Kafka之前，创建Kafka-admin用户

不创建Kafka-admin用户，后续Kafka修改完配置后无法启动，我这里直接使用Zookeeper配置启动，本地可以不启动Kafka，以防止忘记创建用户

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

Kafka所有节点开启SASL认证配置文件，Kafka ACL基于SCRAM-SHA-256

#所有Broker server.properties添加以下配置文件
#advertised.listeners #填写Broker节点IP

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

新增kafka_server_jaas.conf文件

kafka-admin 为管理员用户
kafka-secret 为密码

root@middle-01:/opt/kafka# cat /opt/app/conf/kafka/kafka_server_jaas.conf
KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="kafka-admin"
    password="kafka-secret";
};

添加环境变量，启动后自动引用上面kafka_server_jaas.conf配置

#编辑/opt/kafka/bin/kafka-run-class.sh
vim /opt/kafka/bin/kafka-run-class.sh

#在JMX port to use下面新增配置文件路径
# JMX port to use
if [  $JMX_PORT ]; then
  KAFKA_JMX_OPTS="$KAFKA_JMX_OPTS  -Djava.security.auth.login.config=/opt/app/conf/kafka/kafka_server_jaas.conf"
  if ! echo "$KAFKA_JMX_OPTS" | grep -qF -- '-Dcom.sun.management.jmxremote.rmi.port=' ; then
    # If unset, set the RMI port to address issues with monitoring Kafka running in containers
    KAFKA_JMX_OPTS="$KAFKA_JMX_OPTS -Dcom.sun.management.jmxremote.rmi.port=$JMX_PORT"
  fi
fi

添加完毕后重启Kafka集群，集群内所有Broker都需要添加

root@middle-01:~# ps -ef|grep kafka|awk -F "[ ]+" '{print $2}'|xargs kill -9

#启动Kafka
root@middle-01:~# /opt/kafka/bin/kafka-server-start.sh -daemon /opt/app/conf/kafka/server.properties

#前面为Kafka路径脚本

添加Kafka systemd启动脚本，路径和我这里不一样的请自行修改调整路径

启动脚本路径/opt/kafka/bin/
配置文件路径/opt/app/conf/kafka/server.properties

编辑vim /usr/lib/systemd/system/kafka.service;

[Unit]
Description=Apache Kafka Server
After=network.target

[Service]
Type=forking
User=root
Group=root
Environment="JAVA_HOME=/usr/java/jdk"
ExecStart=/opt/kafka/bin/kafka-server-start.sh -daemon /opt/app/conf/kafka/server.properties
ExecStop=/opt/kafka/bin/kafka-server-stop.sh
Restart=on-failure
TimeoutSec=300

[Install]
WantedBy=multi-user.target

测试脚本是否正常，脚本和进程只可以保留一种方式

systemctl daemon-reload
systemctl start kafka
systemctl enable kafka

启动完毕后查看启动日志

请确保所有Broker节点都启动成功！！！

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

修改kafka-admin用户SASL认证文件,这里我将认证文件放在/ssl/kafka.config,认证协议保持SCRAM-SHA-256

root@middle-01:/opt/kafka# cat /ssl/kafka.config
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="kafka-admin" password="kafka-secret";

创建测试Topic

./bin/kafka-topics.sh --create --bootstrap-server 192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092 \
--topic abcdocker-topic-acl --partitions 3 --replication-factor 2  --command-config /ssl/kafka.config

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

目前Kafka集群配置已经安装完毕，并且开启SASL认证。使用下面的命令查看目前的Topic

root@middle-01:/opt/kafka# ./bin/kafka-topics.sh --bootstrap-server 192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092 --command-config /ssl/kafka.config --list
abcdocker-topic-acl

Kafka 开启ACL配置

开启ACL的前提是要有SASL SCRAM-SHA-256以上的配置，如果没有添加SASL，则无法开启ACL

Kafka Server配置文件新增ACL权限控制

#编辑server.properties
allow.everyone.if.no.acl.found=false
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
super.users=User:kafka-admin

allow.everyone.if.no.acl.found #定义当请求的Topic或操作未找到ACL（访问控制列表）时的默认权限策略,false不允许写入，true代表不用ACL也可以写入
authorizer.class.name=kafka.security.authorizer.AclAuthorizer #启用基于ACL的细粒度权限控制（替代旧版IP白名单）。支持对Topic、Consumer Group、Admin操作等资源的权限管理。
super.users=User:kafka-admin #定义具有超级管理员权限的用户列表

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

添加完毕后重启Kafka服务

systemctl restart kafka

#重启节点如果有数据不一致冲突，只要启动2台节点后，可以手动删除无法启动的data目录，让节点重新启动，对焦数据

重启后请及时验证服务及日志是否正常

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

Kafka节点完整配置如下

参数含义见

中间件Kafka 高可用企业级集群交付文档

24年11月29日
喜欢：0
浏览：286

root@middle-01:/opt/kafka# cat /opt/app/conf/kafka/server.properties
broker.id=1
broker.rack=abcdocker
host.name=middle-01
#log
log.dirs=/data/kafka/kafka-logs
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.check.interval.ms=300000
port=9092
replica.fetch.max.bytes=1000000
#zk
zookeeper.connect=middle-01:2181,middle-02:2181,middle-03:2181
zookeeper.connection.timeout.ms=6000
zookeeper.session.timeout=6000
#sasl
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
sasl.enabled.mechanisms=SCRAM-SHA-256
listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://192.168.21.40:9092
allow.everyone.if.no.acl.found=false
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
super.users=User:kafka-admin

auto.create.topics.enablee=true
compression.type=producer
default.replication.factor=3
delete.topic.enable=true
group.max.session.timeout.ms=1800000
group.min.session.timeout.ms=6000
kafka.scala.version=2.13
log.cleaner.enablee=true
log.cleanup.policy=delete
log.retention.bytes=-1
log.retention.hours=168
log.segment.bytes=1073741824
log.segment.delete.delay.ms=60000
message.max.bytes=1000000
num.io.threads=8
num.network.threads=3
num.partitions=3
num.recovery.threads.per.data.dir=1
num.replica.fetchers=2
offsets.topic.replication.factor=3
queued.max.requests=500
replica.log.time.max.ms=10000
socket.receive.buffer.bytes=102400
socket.send.buffer.bytes=102400
unclean.leader.election.enable=true

创建生产者用户

user-wirte 为生产者用户
222222 为用户的密码
abcdocker-topic-acl 为测试ACL访问权限Topic

创建生产者用户

/opt/kafka/bin/kafka-configs.sh --bootstrap-server 192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092 --command-config /ssl/kafka.config  --alter --add-config 'SCRAM-SHA-256=[password=222222]' --entity-type users --entity-name user-wirte

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

检查用户

/opt/kafka/bin/kafka-configs.sh --bootstrap-server 192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092 --command-config /ssl/kafka.config  --describe --entity-type users

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

创建user-wirte授权文件

vim /ssl/user-wirte.config

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="user-wirte" password="222222";
key.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer
value.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer

使用user-wirte用户消费者测试，无权限写入

/opt/kafka/bin/kafka-console-producer.sh --bootstrap-server 192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092 --producer.config  /ssl/user-wirte.config --topic abcdocker-topic-acl

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

查看abcdocker-topic-acl topic授权用户

[ ] 对于管理topic以及ACL列表查看，需要使用kafka-admin的认证文件

/opt/kafka/bin/kafka-acls.sh --bootstrap-server 192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092  --command-config /ssl/kafka.config --list --topic abcdocker-topic-acl

这里为空

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

创建生产者ACL访问权限

--add #新增添加操作，反过来删除就是--remove
--allow-principal #指定用户
User:user-wirte #User:后面为指定用户
--operation 授权操作
--topic 授权Topic

ACL授权Kafka user-wirte用户对于abcdocker-topic-acl写入操作

/opt/kafka/bin/kafka-acls.sh --bootstrap-server 192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092 --command-config /ssl/kafka.config --add --allow-principal User:user-wirte --operation Write --topic abcdocker-topic-acl

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

查看user-wirte用户授权信息

/opt/kafka/bin/kafka-acls.sh --bootstrap-server 192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092 --command-config /ssl/kafka.config --list --topic abcdocker-topic-acl

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

生产Topic数据测试

/opt/kafka/bin/kafka-console-producer.sh --bootstrap-server 192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092  --producer.config /ssl/user-wirte.config --topic abcdocker-topic-acl

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

消费Topic数据测试，正常情况下user-wirte用户无法消费数据

/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092  --topic abcdocker-topic-acl  --consumer.config /ssl/user-wirte.config  --from-beginning

使用user-wirte用户因为没有权限，会提示下面的报错

root@middle-01:/opt/kafka# /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092  --topic abcdocker-topic-acl  --consumer.config /ssl/user-wirte.config  --from-beginning
[2025-08-23 21:09:04,983] ERROR Error processing message, terminating consumer process:  (org.apache.kafka.tools.consumer.ConsoleConsumer)
org.apache.kafka.common.errors.GroupAuthorizationException: Not authorized to access group: console-consumer-3094
Processed a total of 0 messages

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

创建消费者用户

user-read 为生产者用户
222222 为用户的密码
abcdocker-topic-acl 为测试ACL访问权限Topic
--group * 消费者需要强制指定消费者组，这里我选择*代替

创建消费着user-read用户

/opt/kafka/bin/kafka-configs.sh --bootstrap-server 192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092  --command-config /ssl/kafka.config  --alter --add-config 'SCRAM-SHA-256=[password=222222]' --entity-type users --entity-name user-read

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

创建/ssl/user-read.config

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="user-read" password="222222";
key.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer
value.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer

授权消费者ACL用户权限

授权user-read 读取用户权限(必须指定消费者组)

--consumer #指定消费者参数

/opt/kafka/bin/kafka-acls.sh --bootstrap-server 192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092  --add --allow-principal User:"user-read" --consumer --topic abcdocker-topic-acl --group * --command-config /ssl/kafka.config

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

如果需要移除权限，可以使用--remove （这里不执行）

/opt/kafka/bin/kafka-acls.sh --bootstrap-server  192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092  --remove --allow-principal User:user-read  --operation read --topic abcdocker-topic-acl --group * --command-config  /ssl/kafka.config

查看topic权限列表

/opt/kafka/bin/kafka-acls.sh --bootstrap-server 192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092  --command-config /ssl/kafka.config  --list --topic abcdocker-topic-acl

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

使用user-read消费数据

/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092  --topic abcdocker-topic-acl --consumer.config /ssl/user-read.config  --from-beginning

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

使用user-read用户会发现无法写入数据

/opt/kafka/bin/kafka-console-producer.sh --bootstrap-server 192.168.21.40:9092,192.168.21.41:9092,192.168.21.42:9092 --topic mytest --producer.config /ssl/user-read.config

二进制Kafka_2.13-3.9.0集群开启并配置ACL用户授权

Operation权限定义了用户对资源的具体操作能力，以下是完整的权限分类及说明：

READ

功能：允许消费消息（Consumer操作）
典型场景：消费者读取Topic数据时需此权限

WRITE

功能：允许生产消息（Producer操作）
典型场景：生产者写入Topic数据时需此权限

CREATE

功能：允许创建新资源
适用资源：Topic、Consumer Group等
示例：kafka-topics.sh --create命令需要此权限

DELETE

功能：允许删除资源
风险提示：谨慎授予普通用户该权限

DESCRIBE

功能：允许查看资源元数据
典型命令：kafka-topics.sh --describe
安全建议：避免授予非管理员用户

ALTER

功能：允许修改资源配置
适用场景：调整Topic分区数、副本因子

此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。原文来自 — 版权归原作者所有。

推荐订阅源

i4T

Kafka 开启SASL

Kafka 安全认证SASL

Kafka 开启ACL配置

中间件Kafka 高可用企业级集群交付文档

创建生产者用户

创建生产者ACL访问权限

创建消费者用户

授权消费者ACL用户权限