使用 SASL/PLAIN 认证

1、配置kafka server端（每个broker）

vi $KAFKA_HOME/server.properties

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

在$KAFKA_HOME路径下新建JAAS文件。

vi kafka_server_jaas.conf

KafkaServer {
 org.apache.kafka.common.security.plain.PlainLoginModule required
 username="kafka"
 password="kafka#secret"
 user_kafka="kafka#secret" 
 user_alice="alice#secret"
}

这个配置定义了2个用户（kafka 和 alice）。在KafkaServer部分，username和password是broker用于初始化连接到其他的broker，在这个例子中，kafka用户为broker间的通讯，useruserName定义了所有连接到broker和broker验证的所有的客户端连接包括其他broker的用户密码。（转自链接：http://orchome.com/270） 另，useruserName必须配置kafka用户，否则报错。博客http://wangzzu.github.io/2016/07/29/sasl-plain-kafka/和一些问答中都反应此问题。

JAAS文件作为每个broker的jvm参数，在kafka-server-start.sh脚本中增加如下配置。

vi $KAFKA_HOME/bin/kafka-server-start.sh

if [  "x$KAFKA_OPTS" ]; then
 export KAFKA_OPTS="-Djava.security.auth.login.config=/usr/local/kafka_0.10/kafka_server_jaas.conf"
fi

2、配置kafka client端 PLAIN机制下kafka生产者/消费者如何生产/消费

第一种，console

在$KAFKA_HOME路径下新建JAAS文件。

vi kafka_client_jaas.conf

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

KafkaClient部分，username和password是客户端用来配置客户端连接broker的用户，在这个例子中，客户端使用kafka 用户连接到broker。

修改consuer和producer的配置文件（配置注意空格不然报错）

在$KAFKA_HOME/config/consumer.properties和$KAFKA_HOME/config/producer.properties里分别加上如下配置

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

在启动consumer和producer时，分别添加jvm参数。

vi kafka-console-consumer.sh/kafka-console-producer.sh($KAFKA_HOME/bin目录下)

if [  "x$KAFKA_OPTS" ]; then
 export KAFKA_OPTS="-Djava.security.auth.login.config=/usr/local/kafka_0.10/kafka_client_jaas.conf"
fi

produce

kafka-console-producer.sh --broker-list x.x.x.x:9092 --topic perf --producer.config /usr/local/kafka_0.10/config/producer.properties

consume 只支持新的消费方式--bootstrap-server

不报错 --bootstrap-server

kafka-console-consumer.sh  --bootstrap-server x.x.x.x:9092  --topic perf --from-beginning --consumer.config /usr/local/kafka_0.10/config/consumer.properties --new-consumer

报错 --zookeeper

kafka-console-consumer.sh --zookeeper x.x.x.x:21818/kafka --topic perf  --consumer.config /usr/local/kafka_0.10/config/consumer.properties

第二种，java客户端消费

运行jar包的服务器的指定路径下必须有kafka_client_jaas.conf文件

在程序中添加如下配置

System.setProperty("java.security.auth.login.config", "/tmp/kafka_client_jaas.conf"); //配置文件路径
props.put("security.protocol", "SASL_PLAINTEXT");
props.put("sasl.mechanism", "PLAIN");

本文参考了： kafka使用SASL验证 http://orchome.com/270

Kafka JAAS Plain SASL 安全认证配置 http://blog.csdn.net/geting/article/details/52044055

Kafka 0.10.0 SASL/PLAIN身份认证及权限实现 http://wangzzu.github.io/2016/07/29/sasl-plain-kafka/