1. 配置zookeeper

1. 為zookeeper新增 jaas 檔案

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

2. 在啟動zookeeper時，在KAFKA_OPTS變數里加上" -Djava.security.auth.login.config=/home/work/bin/kafka_2.11-0.10.0.0/sbin/kafka_zoo_jaas.conf"

export KAFKA_OPTS=" -Djava.security.auth.login.config=/home/work/bin/kafka_2.11-0.10.0.0/sbin/kafka_zoo_jaas.conf"
zookeeper-server-start.sh $KAFKA_HOME/config/zookeeper.properties >> zookeeper.log 2>&1 &

2. 配置Server 

1. 為server新增jaas 檔案

KafkaServer {
        org.apache.kafka.common.security.plain.PlainLoginModule required
        username="geting"
        password="geting"
        user_geting="geting"
        user_alice="alice-secret";
};
 

2. 修改server 的配置檔案

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

3. 在啟動server時，在KAFKA_OPTS變數里加上" -Djava.security.auth.login.config=/home/work/bin/kafka_2.11-0.10.0.0/sbin/kafka_server_jaas.conf"

export KAFKA_OPTS=" -Djava.security.auth.login.config=/home/work/bin/kafka_2.11-0.10.0.0/sbin/kafka_server_jaas.conf"
kafka-server-start.sh $KAFKA_HOME/config/server.properties >> kafka.log 2>&1 &

3. 配置client

1.為consuer和producer 新增jaas

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

2. 修改consuer和producer的配置檔案

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

3. 在啟動consumer和producer時，在KAFKA_OPTS變數里加上" -Djava.security.auth.login.config=/home/work/bin/kafka_2.11-0.10.0.0/sbin/kafka_client_jaas.conf"

export KAFKA_OPTS=" -Djava.security.auth.login.config=/home/work/bin/kafka_2.11-0.10.0.0/sbin/kafka_client_jaas.conf"
kafka-console-producer.sh --broker-list localhost:9092 --topic topicName --producer.config $KAFKA_HOME/config/producer.properties

export KAFKA_OPTS=" -Djava.security.auth.login.config=/home/work/bin/kafka_2.11-0.10.0.0/sbin/kafka_client_jaas.conf"
kafka-console-consumer.sh --zookeeper localhost:2181 --bootstrap-server localhost:9092 --topic topicName --from-beginning --consumer.config $KAFKA_HOME/config/consumer.properties --new-consumer