Kafka集群搭建及SASL/SCRAM,ACL权限控制

环境
部署zookeeper集群及安全认证
部署kafka集群及安全认证

环境

系统版本：Centos7.4
kafka版本：kafka_2.12-2.6.0
zookeeper版本：zookeeper-3.6.2
JDK版本：jdk1.8

服务器：
192.168.18.11
192.168.18.12
192.168.18.13

部署目录
/webapps/kafka
/webapps/zookeeper

部署zookeeper集群及安全认证

zookeeper集群,就是配置文件中相互加入
1.修改配置文件

3台相同
vi /webapps/zookeeper/conf/zoo.cfg

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/webapps/zookeeper_data/
clientPort=12181
server.1=192.168.18.11:12888:13888
server.2=192.168.18.12:12888:13888
server.3=192.168.18.13:12888:13888
##SASL安全认证----
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000
##------

2.每台服务器增加唯一myid
上一步配置文件中目录 dataDir=/webapps/zookeeper_data/

#192.168.18.11配置
cd /webapps/zookeeper_data
echo 1 > myid

#192.168.18.12配置
cd /webapps/zookeeper_data
echo 2 > myid

#192.168.18.13配置
cd /webapps/zookeeper_data
echo 3 > myid

3.增加安全认证信息文件zk_server_jaas.conf

3台服务器都相同,conf文件可放在任意位置
vi /webapps/zookeeper/conf/zk_server_jaas.conf

Server {org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin" user_admin="admin"user_kafka="kafka";
};

4.将kafka相关jar包导入zookeeper
将kafka/lib下相关jar包复制到zookeeper/lib下

kafka-clients-1.1.1.jar
lz4-java-1.4.1.jar
snappy-java-1.1.7.1.jar

5.修改bin/zkEnv.sh,让zk启动时读取到zk_server_jaas.conf

vi /webapps/zookeeper/bin/zkEnv.sh#增加
export SERVER_JVMFLAGS=" -Djava.security.auth.login.config=/webapps/zookeeper/conf/zk_server_jaas.conf"

6.启动
bin/zkServer.sh start

其他命令
停止:bin/zkServer.sh stop
重启:bin/zkServer.sh restart
状态:bin/zkServer.sh status

部署kafka集群及安全认证

1.增加安全认证信息文件kafka_server_jaas.conf
3台服务器相同
KafkaServer 块中,是kafka设置的认证信息
Client 块中,是zookeeper的认证信息,用于kafka连接zookeeper

vi /webapps/kafka/config/kafka_server_jaas.conf

KafkaServer {org.apache.kafka.common.security.scram.ScramLoginModule requiredusername="admin"password="admin";
};Client {org.apache.kafka.common.security.plain.PlainLoginModule requiredusername="kafka"password="kafka";
};

2.创建认证
设置config启动变量
vi /webapps/kafka/bin/kafka-configs.sh

增加

export KAFKA_HEAP_OPTS="-Djava.security.auth.login.config=/webapps/kafka/config/kafka_server_jaas.conf"

#创建admin认证,password对应kafka_server_jaas.conf
bin/kafka-configs.sh --zookeeper 192.168.18.11:12181 --alter --add-config 'SCRAM-SHA-256=[password=admin],SCRAM-SHA-512=[password=admin]' --entity-type users --entity-name admin#查看证书
bin/kafka-configs.sh --zookeeper 192.168.18.11:12181 --describe --entity-type users  --entity-name admin#删除证书(不操作)
bin/kafka-configs.sh --zookeeper 192.168.18.11:12181 --alter --delete-config 'SCRAM-SHA-256' --entity-type users --entity-name admin

3.修改配置文件
vi /webapps/kafka/config/server.properties

#192.168.18.11配置
#设置集群内唯一标识
broker.id=1listeners=SASL_PLAINTEXT://0.0.0.0:9092#安全认证监控服务
advertised.listeners=SASL_PLAINTEXT://192.168.18.11:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
allow.everyone.if.no.acl.found=true
#设置超级用户
super.users=User:admin#关闭自动创建topic,默认true
auto.create.topics.enable=false#log目录
log.dirs=/webapps/kafka/logs#分区数
num.partitions=3#topic的offset的备份份数
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
#默认备份份数，仅指自动创建的topics
default.replication.factor=1#zookeeper连接串
zookeeper.connect=192.168.18.11:12181,192.168.18.12:12181,192.168.18.13:12181

#192.168.18.12配置#设置集群内唯一标识
broker.id=2listeners=SASL_PLAINTEXT://0.0.0.0:9092#安全认证监控服务
advertised.listeners=SASL_PLAINTEXT://192.168.18.12:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
allow.everyone.if.no.acl.found=true
#设置超级用户
super.users=User:admin#关闭自动创建topic,默认true
auto.create.topics.enable=false#log目录
log.dirs=/webapps/kafka/logs#分区数
num.partitions=3#topic的offset的备份份数
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
#默认备份份数，仅指自动创建的topics
default.replication.factor=1#zookeeper连接串
zookeeper.connect=192.168.18.11:12181,192.168.18.12:12181,192.168.18.13:12181

#192.168.18.13配置#设置集群内唯一标识
broker.id=3listeners=SASL_PLAINTEXT://0.0.0.0:9092#安全认证监控服务
advertised.listeners=SASL_PLAINTEXT://192.168.18.13:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
allow.everyone.if.no.acl.found=true
#设置超级用户
super.users=User:admin#关闭自动创建topic,默认true
auto.create.topics.enable=false#log目录
log.dirs=/webapps/kafka/logs#分区数
num.partitions=3#topic的offset的备份份数
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
#默认备份份数，仅指自动创建的topics
default.replication.factor=1#zookeeper连接串
zookeeper.connect=192.168.18.11:12181,192.168.18.12:12181,192.168.18.13:12181

4.配置kafka启动环境变量
vi /webapps/kafka/bin/kafka-server-start.sh

#增加
export KAFKA_HEAP_OPTS="-Xmx2G -Xms2G -Djava.security.auth.login.config=/webapps/kafka/config/kafka_server_jaas.conf"

5.启动
bin/kafka-server-start.sh -daemon config/server.properties

6.增加admin认证文件
vi /webapps/kafka/config/admin.conf

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

7.创建topic

#创建topic
bin/kafka-topics.sh --create  --bootstrap-server  192.168.18.11:9092 --replication-factor 1 --partitions 1 --topic mytest --command-config config/admin.conf#查看所有topic
bin/kafka-topics.sh --list  --bootstrap-server  192.168.18.11:9092  --command-config config/admin.conf

8.操作用户

#新增用户,新建用户mytest
bin/kafka-configs.sh  --zookeeper 192.168.18.11:12181 --alter --add-config 'SCRAM-SHA-512=[password=123456]' --entity-type users --entity-name mytest#更新用户,更新mytest的密码为mytest
bin/kafka-configs.sh --zookeeper 192.168.18.11:12181 --alter --add-config 'SCRAM-SHA-512=[password=mytest]' --entity-type users --entity-name mytest

9.设置权限

#读取权限,设置用户mytest的消费者权限
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.18.11:12181 --add --allow-principal User:"mytest" --consumer --topic 'mytest' --group '*'#写入权限,设置用户mytest的生产者权限
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.18.11:12181 --add --allow-principal User:"mytest" --producer --topic 'mytest'#查看权限
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.18.11:12181 --list

10.测试

新建mytest认证文件

vi /webapps/kafka/config/mytest.conf

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

2台机器分别执行生产者和消费者-------------生产者生产消息,看消费者是否能接收到

#生产者
bin/kafka-console-producer.sh --broker-list 192.168.18.11:9092 --topic mytest --producer.config config/mytest.conf#消费者
bin/kafka-console-consumer.sh --bootstrap-server 192.168.18.11:9092 --topic mytest --consumer.config config/mytest.conf

11.查看groupid和offset

#查看所有groupid
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.18.11:9092 --list  --command-config config/admin.conf#查看offset
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.18.11:9092 --describe --group  test-consumer-group --command-config config/admin.conf

12.查看消息内容

#查看消息内容: 消息追踪
bin/kafka-run-class.sh kafka.tools.DumpLogSegments  --files /webapps/kafka/logs/mytest-0/00000000000000000000.log --print-data-log