SASL_PLAINTEXT 實現認證

1. server.properties中新增：

listeners=SASL_PLAINTEXT://hostname:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN

2. config目錄下新建三個檔案：

kafka_server_jaas.conf

KafkaServer {
        org.apache.kafka.common.security.plain.PlainLoginModule required
        username="admin"
        password="admin-pwd"
        user_admin="admin-pwd"
        user_alice="alice-pwd"
        user_bob="bob-pwd";
};
 

kafka_cilent_jaas.conf

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

kafka_zoo_jaas.conf

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

3. 修改kafka各項sh指令碼 
zookeeper-server-start.sh中新增（要在最後一行之前新增，別新增在最後一行了）：

export KAFKA_OPTS=" -Djava.security.auth.login.config=/home/ocdc/app/kafka_2.10-0.10.1.1/config/kafka_zoo_jaas.conf"

kafka-server-start.sh中新增：

export KAFKA_OPTS=" -Djava.security.auth.login.config=/home/ocdc/app/kafka_2.10-0.10.1.1/config/kafka_server_jaas.conf"

kafka-console-consumer.sh和kafka-console-producer.sh中新增：

export KAFKA_OPTS=" -Djava.security.auth.login.config=/home/ocdc/app/kafka_2.10-0.10.1.1/config/kafka_client_jaas.conf"

4.啟動zookeeper 和Kafka叢集

上述步驟，為叢集添加了SASL_PLAINTEXT認證。Kafka Server間/Kafka和ZK間，利用admin賬號互聯。

同時Kafka Server還保留了兩個普通使用者：alice和bob，密碼分別為alice-pwd,bob-pwd。

5.測試

利用kafka-console-consumer.sh和kafka-console-producer.sh，同時修改kafka_cilent_jaas.conf中的使用者名稱和密碼，測試認證機制是否生效。

Kafka-ACL實現許可權控制

1. server.properties中新增：
authorizer.class.name = kafka.security.auth.SimpleAclAuthorizer
super.users=User:admin
Kafka Server間是使用admin賬號互聯的，使用得把admin配為超級賬號 2. 重啟Kafka叢集 3.利用kafka-acls.sh為topic設定ACL sh kafka-acls.sh --authorizer-properties zookeeper.connect=hostname:2181 --add --allow-principal  User:alice  --group test-consumer-group --topic alice-topic
這裡注意，如果alice要作為消費端連線alice-topic的話，必須對其使用的group（test-consumer-group）也賦權

4. 利用kafka-acls.sh設定不同的許可權，同時修改kafka_cilent_jaas.conf中的使用者名稱和密碼，驗證ACL和認證機制。