Kafka SASL集群部署

Kafka SASL集群部署

在Kafka中，SASL機制包含三種，它們分別是Kerberos、PLAIN、SCRAM，以PLAIN認證為示例

1.配置Server

1）解壓安裝包

tar -zxvf kafka_2.11-2.4.1.tgz -C /home/xyp9x/

2）改名

mv kafka_2.11-2.4.1 kafka_sasl

2）在kafka目錄下創建logs、kafka-logs檔案夾

mkdir logs kafka-logs

3）config目錄中創建kafka_server_jaas.conf檔案，前三行是配置管理員賬戶（該賬戶與上面server.properties中配置的super.users一樣）后面的即 user_用戶名=“該用戶的密碼”，之后配置ACL的時候需要用到這里配置的用戶

vi kafka_server_jaas.conf

KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin"
user_admin="admin"
user_xyp9x="xyp9x";
};

4）修改組態檔

vi server.properties

將原內容全部清空，輸入以下內容：
#broker的全域唯一編號，不能重復
broker.id=0
#是否允許洗掉topic
delete.topic.enable=true
#處理網路請求的執行緒數量
num.network.threads=3
#用來處理磁盤IO的執行緒數量
num.io.threads=8
#發送套接字的緩沖區大小
socket.send.buffer.bytes=102400
#接收套接字的緩沖區大小
socket.receive.buffer.bytes=102400
#請求套接字的最大緩沖區大小
socket.request.max.bytes=104857600
#kafka資料存放的路徑
log.dirs=/home/xyp9x/kafka_sasl/kafka-logs
#topic在當前broker上的磁區個數
num.partitions=1
#用來恢復和清理data下資料的執行緒數量
num.recovery.threads.per.data.dir=1
#segment檔案保留的最長時間，超時將被洗掉
log.retention.hours=168
#配置連接Zookeeper集群地址
zookeeper.connect=bigdata111:2181,bigdata112:2181,bigdata113:2181
#kafka連接zookeeper超時時間90s
zookeeper.connection.timeout.ms=90000

#SASL_PLAINTEXT 
#這里的listener中的hostname在三臺機器上換成每臺機器對應的hostname:ip
listeners=SASL_PLAINTEXT://192.168.1.111:9092
#使用的認證協議
security.inter.broker.protocol=SASL_PLAINTEXT
#SASL機制
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
#完成身份驗證的類
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
#如果沒有找到ACL（訪問控制串列）配置，則允許任何操作，
#設定為true，ACL機制為黑名單機制，只有黑名單中的用戶無法訪問
#默認為false，ACL機制為白名單機制，只有白名單中的用戶可以訪問
#allow.everyone.if.no.acl.found=true
#設定admin超級用戶
super.users=User:admin

5）然后在Kafka啟動腳本中添加組態檔路徑

vi kafka-server-start.sh

#!/bin/bash
export KAFKA_OPTS="-Djava.security.auth.login.config=/home/xyp9x/kafka_sasl/config/kafka_server_jaas.conf"

6）配置環境變數

vi /etc/profile

# Kafka_sasl
export KAFKA_HOME=/home/xyp9x/kafka_sasl
export PATH=$PATH:$KAFKA_HOME/bin

7）分發

rsync -r /etc/profile bigdata112:/etc/
rsync -r /etc/profile bigdata113:/etc/
rsync -r kafka_sasl bigdata112:`pwd`
rsync -r kafka_sasl bigdata113:`pwd`

8）分別在112和113上修改組態檔

vi server.properties
broker.id=1
listeners=SASL_PLAINTEXT://192.168.1.112:9092

vi server.properties
broker.id=2
listeners=SASL_PLAINTEXT://192.168.1.113:9092

9）更新111、112、113的環境變數

source /etc/profile

2.配置Client（當Kafka Server端配置啟用了SASL/PLAIN，那么Client連接的時候需要配置認證資訊）

1）在config目錄創建kafka_client_jaas.conf

vi kafka_client_jaas.conf

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

2）在producer.properties和consumer.properties檔案中設定認證協議

vi producer.properties

security.protocol=SASL_PLAINTEXT 
sasl.mechanism=PLAIN

vi consumer.properties

security.protocol=SASL_PLAINTEXT 
sasl.mechanism=PLAIN

3）在kafka-console-producer.sh腳本和kafka-console-consumer.sh腳本中添加JAAS檔案的路徑

vi kafka-console-producer.sh

#!/bin/bash
export KAFKA_OPTS="-Djava.security.auth.login.config=/home/xyp9x/kafka_sasl/config/kafka_client_jaas.conf"

vi kafka-console-consumer.sh

#!/bin/bash
export KAFKA_OPTS="-Djava.security.auth.login.config=/home/xyp9x/kafka_sasl/config/kafka_client_jaas.conf"

4）分發

rsync -r config/kafka_client_jaas.conf bigdata112:/home/xyp9x/kafka_sasl/config/
rsync -r config/kafka_client_jaas.conf bigdata113:/home/xyp9x/kafka_sasl/config/
rsync -r config/producer.properties bigdata112:/home/xyp9x/kafka_sasl/config/
rsync -r config/producer.properties bigdata113:/home/xyp9x/kafka_sasl/config/
rsync -r config/consumer.properties bigdata112:/home/xyp9x/kafka_sasl/config/
rsync -r config/consumer.properties bigdata113:/home/xyp9x/kafka_sasl/config/
rsync -r bin/kafka-console-producer.sh bigdata112:/home/xyp9x/kafka_sasl/bin/
rsync -r bin/kafka-console-producer.sh bigdata113:/home/xyp9x/kafka_sasl/bin/
rsync -r bin/kafka-console-consumer.sh bigdata112:/home/xyp9x/kafka_sasl/bin/
rsync -r bin/kafka-console-consumer.sh bigdata113:/home/xyp9x/kafka_sasl/bin/

5）以安全認證的方式啟動 kafka-server：

bin/kafka-server-start.sh -daemon config/server.properties

6）創建一個 topic：

bin/kafka-topics.sh --create --zookeeper 192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --replication-factor 3 --partitions 3 --topic test

7）以安全認證的方式啟動 kafka-producer

bin/kafka-console-producer.sh --broker-list 192.168.1.111:9092,192.168.1.112:9092,192.168.1.113:9092 --topic test --producer.config config/producer.properties

8）以安全認證的方式啟動 kafka-consumer

bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.111:9092,192.168.1.112:9092,192.168.1.113:9092 --topic test --consumer.config config/consumer.properties

3.ACL操作（在配置好SASL后，啟動Zookeeper集群和Kafka集群之后，就可以使用kafka-acls.sh腳本來操作ACL機制）

1）查看：在kafka-acls.sh腳本中傳入list引數來查看ACL授權

bin/kafka-acls.sh --list --authorizer-properties zookeeper.connect=192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181

2）創建：創建待授權主題之前，在kafka-acls.sh腳本中指定JAAS檔案路徑，然后在執行創建操作

bin/kafka-topics.sh --create --zookeeper 192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --replication-factor 3 --partitions 3 --topic kafka_acl_topic

3）生產者授權：對生產者執行授權操作

bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --add --allow-principal User:producer --operation Write --topic kafka_acl_topic

4）消費者授權：對生產者執行授權后，通過消費者來進行驗證

bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --add --allow-principal User:consumer --operation Read --topic kafka_acl_topic

5）洗掉：通過remove引數來回收生產者權限

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --remove --allow-principal User:producer --operation Write --topic kafka_acl_topic

6）洗掉：通過remove引數來回收消費者權限

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --remove --allow-principal User:consumer --operation Read --topic kafka_acl_topic