- 创建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";
};
几点说明：
- username/password是给kafka brokers之间作为client初始请求连接访问使用的，会被作为server的broker验证。
- user_kafkaadmin这个就是1中提到的连接访问请求验证信息，所以这条是必须的。
- user_kafkaclient1/user_kafkaclient2定义了kafka的client，其值就是密码。
- Client是把kafka作为client端，访问zookeeper(作为server端)的时候用的。对应的必须在zookeeper里面做相应的配置。
这个Client是可选的，如果zookeeper没有配置SASL认证属性，则可以省略；见下面。
另外要吐槽这个文件的格式，好神奇啊，为什么最后一项需要分号呢？？？
- 把上述配置文件传递给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"
- 修改kafka配置server.properties
listeners=SASL_PLAINTEXT://:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
- 为zookeeper创建配置文件
$ cat zookeeper_jaas.conf
Server {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="zooadmin"
password="zooadminpwd"
user_zooclient="zooclientpwd";
};
几点说明：
- username/password我还不知道是干嘛用的。
- user_zooclient这个用户就是前面kafka_server_jaas.conf里面的Client信息；这样就对上了：这两个用户名和密码必须一直，作为kafka client和zookeeper server之间认证使用。
- 把上述配置文件传递给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数据。
注意针对4和5，像在1里面提到的，如果不需要配置zookeeper的SASL认证功能，那么4和5都可以忽略，也就是说kafka访问zookeeper的时候不需要SASL认证；此时kafka_server_jaas.conf里面的Client部分可以省略，这样你在启动kafka的时候会看到如下类型信息：
[2018-12-26 09:33:22,220] WARN SASL configuration failed:
javax.security.auth.login.LoginException: No JAAS configuration section named 'Client'
was found in specified JAAS configuration file:
'/opt/kafka/config/kafka_server_jaas.conf'.
Will continue connection to Zookeeper server without SASL authentication,
if Zookeeper server allows it. (org.apache.zookeeper.ClientCnxn)
这就依赖zookeeper的配置觉得kafka能不能启动成功。
- 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。
- 最严重问题
这种认证模式，账号信息都是提前写在kafka_server_jaas.conf文件里面的，不能做到动态修改；例如新增账号，删除帐号，修改密码等等不能在线操作，必须重启kafka，这导致这种认证模式根本无法在产品线上运行，只能作为测试，或者小范围内使用，因为不能停机重启啊。
