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"
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"
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/