用Cloudera Manager 安裝 的kafka默認情況下，可以開啟 Kerberos 和Sentry做安全認證和授權(quán)服務(wù)畏吓。
但有時有些特殊需求墨状，不適合在Kerberos中添加用戶，可以通過添加其他認證機制來解決此問題菲饼。
本文講述了 如何讓 Cloudera Manager中的kafka 支持 PLAIN和 SCRAM認證機制
版本：CDH-->5.13, Cloudera Manager-->5.13, kafka-->3.1.0對應(yīng)的社區(qū)版：1.0.1

Step 1：登錄 Cloudera Manager 控制臺肾砂，進入kafka 組件的configuration 中。

Step 2：修改如下選項

Sentry Service：none 說明：關(guān)閉Sentry宏悦，因為SCRAM和PLAIN的授權(quán)操作都是通過kafka-acl來完成的镐确，授權(quán)信息存儲在zookeeper中，而Sentry 是獨立的權(quán)限控制系統(tǒng)饼煞，只能支持Kerberos源葫，不能支持SCRAM和PLAIN。

advertised.host.name：說明：配置每個broker的IP地址或者主機名

advertised.port：9093 說明：此示例中開啟了PLAINTEXT-->9092和SASL_PLAINTEXT-->9093端口

security.inter.broker.protocol：INFERRED 說明：自動推測協(xié)議類型砖瞧，推測結(jié)果為SASL_PLAINTEXT

authenticate.zookeeper.connection：true 連接 zookeeper是 啟用SASL協(xié)議息堂，因為 zookeeper開啟了kerberos認證，因此此處需要勾選块促。

super.users：kafka 啟動kafka服務(wù)的 超級用戶荣堰。默認為kafka用戶

broker_java_opts：尾部增加 -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf，指定kafka broker啟動是 加載本地jaas文件/etc/kafka/kafka_server_jaas.conf竭翠，后面需要為每個機器設(shè)置kafka_server_jaas.conf文件

Kafka Broker Advanced Configuration Snippet (Safety Valve) for kafka.properties：內(nèi)容如下
[
listeners=PLAINTEXT://0.0.0.0:9092,SASL_PLAINTEXT://0.0.0.0:9093
sasl.enabled.mechanisms=GSSAPI,SCRAM-SHA-256,PLAIN
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=GSSAPI
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
]
另外在每個broker上的此項配置中再增加如下信息
[
advertised.listeners=PLAINTEXT://hostname:9092,SASL_PLAINTEXT://hostname:9093
]

Step 3: 在每個broker主機上持隧，添加/etc/kafka/kafka_server_jaas.conf，內(nèi)容如下：

KafkaServer {
org.apache.kafka.common.security.scram.ScramLoginModule required
   username=kafka
   password=thisIsSecret;
org.apache.kafka.common.security.plain.PlainLoginModule required
   username=kafka
   password=thisIsSecret
   user_kafka=thisIsSecret
   user_alice=thisIsSecret;
com.sun.security.auth.module.Krb5LoginModule required
   doNotPrompt=true
   useKeyTab=true
   storeKey=true
   useTicketCache=true
   keyTab="/etc/kafka/kafka.keytab"
   principal="kafka/hostname@example.com";
};
KafkaClient {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   storeKey=true
   useTicketCache=true
   keyTab="/etc/kafka/kafka.keytab"
   principal="kafka/hostname@example.com";
};
Client {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   storeKey=true
   useTicketCache=true
   keyTab="/etc/kafka/kafka.keytab"
   principal="kafka/hostname@example.com";
};

配置項中的principal 改為 實際的kerberos中的principal逃片，及對應(yīng)的keytab文件屡拨。
至此 可以通過PLAIN 和Kerberos 兩種認證機制來訪問kafka只酥。
如果需要使用SCRAM機制，可以通過如下命令添加kafka和alice用戶

/opt/cloudera/parcels/KAFKA-3.0.0-1.3.0.0.p0.40/lib/kafka/bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[password=thisIsSecret]' --entity-type users --entity-name kafka
/opt/cloudera/parcels/KAFKA-3.0.0-1.3.0.0.p0.40/lib/kafka/bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[password=thisIsSecret]' --entity-type users --entity-name alice

至此可以通過PLAIN,KERBEROS,SCRAM三種認證機制來訪問kafka集群呀狼。

創(chuàng)建topic：

kafka-topics --zookeeper localhost:2181--list 
kafka-topics --create --zookeeper localhost:2181--partitions 30 --replication-factor 1 --topic test

授權(quán)：

/opt/cloudera/parcels/KAFKA-3.0.0-1.3.0.0.p0.40/lib/kafka/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --operation IdempotentWrite --allow-principal User:* --allow-host=* --cluster
/opt/cloudera/parcels/KAFKA-3.0.0-1.3.0.0.p0.40/lib/kafka/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --operation All --allow-principal User:* --allow-host=* --transactional-id=*
/opt/cloudera/parcels/KAFKA-3.0.0-1.3.0.0.p0.40/lib/kafka/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --operation All --allow-principal User:* --allow-host=* --cluster
/opt/cloudera/parcels/KAFKA-3.0.0-1.3.0.0.p0.40/lib/kafka/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:* --allow-host=* --operation All --group=*  --topic test
/opt/cloudera/parcels/KAFKA-3.0.0-1.3.0.0.p0.40/lib/kafka/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:* --allow-host=* --operation All --group=*  --topic __consumer_offsets

客戶端訪問：
在客戶端 創(chuàng)建如下plain_alice_jaas.conf 文件

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

scram_alice_jaas.conf

KafkaClient {
org.apache.kafka.common.security.scram.ScramLoginModule required
username=alice
password=thisIsSecret;
};

啟動PLAIN機制的producer

export KAFKA_OPTS=" -Djava.security.auth.login.config=`pwd`/plain_alice_jaas.conf"
kafka-console-producer --broker-list localhost:9093 --producer-property security.protocol=SASL_PLAINTEXT --producer-property sasl.mechanism=PLAIN --topic test

啟動SCRAM機制的consumer

export KAFKA_OPTS=" -Djava.security.auth.login.config=`pwd`/scram_alice_jaas.conf"
kafka-console-consumer --new-consumer --bootstrap-server localhost:9093 --consumer-property security.protocol=SASL_PLAINTEXT --consumer-property sasl.mechanism=SCRAM-SHA-256 --topic test

Kerberos認知機制如下
krb5_alice_jaas.conf文件

KafkaClient {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   storeKey=true
   useTicketCache=true
   keyTab="/etc/kafka/alice.keytab"
   principal="alice/hostname@example.com";
};
export KAFKA_OPTS=" -Djava.security.auth.login.config=`pwd`/krb5_alice_jaas.conf"
kafka-console-consumer --new-consumer --bootstrap-server localhost:9093 --consumer-property security.protocol=SASL_PLAINTEXT --consumer-property sasl.mechanism=GSSAPI --topic test