Kafka添加Acl認(rèn)證

一、Kafka安裝

#### 1. 在安裝Kafka的基礎(chǔ)配置上添加或修改以下參數(shù)
$KAFKA_HOME/config/server.properties
listeners=SASL_PLAINTEXT://hadoop02:9092

如果設(shè)置此參數(shù)在集群狀態(tài)下啟動會失敗

allow.everyone.if.no.acl.found=true
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

超級用戶

super.users=User:admin
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

2. 修改Kafka啟動腳本添加以下內(nèi)容

$KAFKA_HOME/bin/kafka-server-start.s
export KAFKA_HEAP_OPTS="$KAFKA_HEAP_OPTS -Djava.security.auth.login.config=$KAFKA_HOME/jass/kafka-server-jaas.conf"

3. 創(chuàng)建kafka-server-jass.conf配置文件

$KAFKA_HOME/jass/kafka-server-jaas.conf

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin"
    user_admin="admin"
    user_writer="writer"
    user_reader="reader";
};

此文件做認(rèn)證使用 格式為： user_用戶=密碼

二、配置Producter（生產(chǎn)者）

1. 修改Kafka生產(chǎn)者腳本 添加以下內(nèi)容

$KAFKA_HOME/bin/kafka-console-producer.sh

KAFKA_HEAP_OPTS="$KAFKA_HEAP_OPTS -Djava.security.auth.login.config=$KAFKA_HOME/jass/kafka-writer-jass.conf"

2. 創(chuàng)建生產(chǎn)者jass文件

$KAFKA_HOME/jass/kafka-writer-jass.conf
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="writer"
password="writer";
};

image.png

3. 創(chuàng)建生產(chǎn)者腳本啟動需要的配置文件

$KAFKA_HOME/jass/producer.conf
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

三、創(chuàng)建測試主題測試生產(chǎn)者

1. 創(chuàng)建wxgz主題

kafka-topics.sh --create --zookeeper hadoop02:2181 --topic wxgz --partitions 1 --replication-factor 1

創(chuàng)建成功（注意正常來說加了驗證是不應(yīng)該創(chuàng)建成功的，但是kafka-topic.sh腳本是直接和zookeeper交互的，
不經(jīng)過kafka所以創(chuàng)建成功，也就是說kafka-topic.sh腳本不受acl的控制）

2. 給用戶授權(quán)

kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=hadoop02:2181 --add --allow-principal User:writer --operation Write --topic wxgz

3. 測試 向wxgz 主題中生產(chǎn)數(shù)據(jù)

kafka-console-producer.sh --broker-list hadoop02:9092 --topic wxgz --producer.config jass/producer.conf

寫入成功

四、配置和測試消費者

1. 修改消費者腳本 添加一下內(nèi)容

$KAFKA_HOME/bin/kafka-console-consumer.sh
KAFKA_HEAP_OPTS="$KAFKA_HEAP_OPTS -Djava.security.auth.login.config=$KAFKA_HOME/jass/kafka-reader-jass.conf"

2. 創(chuàng)建消費者jass配置文件

$KAFKA_HOME/jass/kafka-reader-jass.confsecurity.protocol=SASL_PLAINTEXT
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="reader"
password="reader";
};

3. 創(chuàng)建消費者腳本啟動需要的配置文件

$KAFKA_HOME/jass/consumer.conf
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
group.id=test-group

4. 授予用戶讀權(quán)限

kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=hadoop02:2181 --add --allow-principal User:reader --operation Read --topic wxgz

5. 測試消費數(shù)據(jù)

kafka-console-consumer.sh --bootstrap-server hadoop02:9092 --topic wxgz --from-beginning --consumer.config jass/consumer.conf

五、常用命令

#### 1.創(chuàng)建主題
kafka-topics.sh --create --zookeeper hadoop02:2181 --topic wxgz --partitions 1 --replication-factor 1

#### 2. 授予用戶寫權(quán)限
kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=hadoop02:2181 --add --allow-principal User:writer --operation Write --topic wxgz

#### 3. 啟動生產(chǎn)者
kafka-console-producer.sh --broker-list hadoop02:9092 --topic wxgz --producer.config jass/producer.conf

#### 4. 授予用戶讀權(quán)限
kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=hadoop02:2181 --add --allow-principal User:reader --operation Read --topic wxgz

#### 5. 授予消費組讀權(quán)限
kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=hadoop02:2181 --add --allow-principal User:reader --operation Read --group test-group

#### 6. 啟動消費者
kafka-console-consumer.sh --bootstrap-server hadoop02:9092 --topic wxgz --from-beginning --consumer.config jass/consume.conf