kafka如何配置SASL/PLAIN认证模式

1. 创建kafka broker配置文件

$ cat kafka_server_jaas.conf 
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="kafkaadmin"
    password="kafkaadminpwd"
    user_kafkaadmin="kafkaadminpwd"
    user_kafkaclient1="kafkaclient1pwd"
    user_kafkaclient2="kafkaclient2pwd";
};

Client {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="zooclient"
    password="zooclientpwd";
};

几点说明：

1. username/password是给kafka brokers之间作为client初始请求连接访问使用的，会被作为server的broker验证。
2. user_kafkaadmin这个就是1中提到的连接访问请求验证信息，所以这条是必须的。
3. user_kafkaclient1/user_kafkaclient2定义了kafka的client，其值就是密码。
4. Client是把kafka作为client端，访问zookeeper(作为server端)的时候用的。对应的必须在zookeeper里面做相应的配置。

另外要吐槽这个文件的格式，好神奇啊，为什么最后一项需要分号呢？？？

2. 把上述配置文件传递给kafka

作为JVM的参数。

-Djava.security.auth.login.config=/path/to/kafka_client_jaas.conf

具体操作可以通过环境变量KAFKA_OPTS来传递，如果用户使用/opt/kafka/bin/kafka-server-start.sh来启动kafka。

export KAFKA_OPTS="-Djava.security.auth.login.config=/path/to/kafka_client_jaas.conf"

3. 修改kafka配置server.properties

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

4. 为zookeeper创建配置文件

$ cat zookeeper_jaas.conf
Server {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="zooadmin"
    password="zooadminpwd"
    user_zooclient="zooclientpwd";
};

几点说明：

1. username/password我还不知道是干嘛用的。
2. user_zooclient这个用户就是前面kafka_server_jaas.conf里面的Client信息；这样就对上了：这两个用户名和密码必须一直，作为kafka client和zookeeper server之间认证使用。

5. 把上述配置文件传递给zookeeper

作为JVM的参数。

-Djava.security.auth.login.config=/path/to/zookeeper_jaas.conf

具体操作也可以通过环境变量KAFKA_OPTS来传递(注意这里也是KAFKA_OPTS，而不是ZOOKEEPER_OPTS，或者ZK_OPTS)，如果用户使用/opt/kafka/bin/zookeeper-server-start.sh来启动kafka。

export KAFKA_OPTS="-Djava.security.auth.login.config=/path/to/zookeeper_jaas.conf"

完毕，这样kafka端就配置完成了。下面我们用goclient来访问这个kafka数据。

6. golang的客户端

我没有写完整的golang客户端代码，只把其中相关的放在这儿。

config := sarama.NewConfig()

config.Net.SASL.Enable = true
config.Net.SASL.User = "kafkaclient1"
config.Net.SASL.Password = "kafkaclient1pwd"

client, err := sarama.NewClient(strings.Split(hosts, ","), config)
if err != nil {
    log.Fatalf("unable to create kafka client: %q", err)
}

consumer, err := sarama.NewConsumerFromClient(client)
defer consumer.Close()
//producer, err := sarama.NewSyncProducerFromClient(client)
//defer producer.Close()

// handle consumer/producer

这里，配置了访问kafka的用户信息：

config.Net.SASL.Enable = true
config.Net.SASL.User = "kafkaclient1"
config.Net.SASL.Password = "kafkaclient1pwd"

如果账号填写错误NewClient就会失败，返回错误：

2018/12/25 06:45:39 unable to create kafka client: 
"kafka: client has run out of available brokers to talk to (Is your cluster reachable?)"

这个错误信息倒是很晕，根本不说是用户连接信息错误，只说这么模糊的错误，调试起来费老大劲了。

注意目前Shopify/sarama对SASL的支持只有PLAINTEXT模式，其他的SCRAM，GSSAPI都不支持。所有严格安全性来说还是需要用SSL。

7. 最严重问题

这种认证模式，账号信息都是提前写在kafka_server_jaas.conf文件里面的，不能做到动态修改；例如新增账号，删除帐号，修改密码等等不能在线操作，必须重启kafka，这导致这种认证模式根本无法在产品线上运行，只能作为测试，或者小范围内使用，因为不能停机重启啊。