背景

项目需求是：通过 Web 页面配置 Kafka 信息并存储，且支持安全认证模式。

本文介绍 Kafka 帐号密码认证的完整实践流程，自己实践一遍，才能穿起各个概念。Kafka 的 Plain 简单文本认证方式比较简单，只需要 Kafka 服务端维护用户列表，客户端同样的安全认证配置即可。

本文的实践过程中，Kafka 版本为 kafka_2.11-2.3.1 ，只对 Kafka 服务做安全认证；Zookeeper 使用自带的，不做安全认证。

关键配置列表：

1. Kafka 服务配置文件 server.propertis，配置认证协议及认证实现类；
2. Kafka 的 jaas.config 认证配置文件，登录类，超管密码和管理的帐号密码列表；
3. Kafka 服务启动脚本 kafka-start-server.sh，设置安全认证环境变量；
4. Kafka 客户端连接配置认证属性。

0、知识预备，了解几个概念

首先，需要了解Kafka 有三种认证模式：Kerberos、Plain【简单帐号密码模式】、SSL 模式。

其次，Plain，简单帐号密码模式。它又分为 SASL/SCRAM 和 SASL/PLAIN 两种方式。区别是：SASL/SCRAM 可以在 Kafka 服务启动之后，动态的新增用户分并配权限，在业务变动频繁，开发人员多的情况下比 SASL/PLAIN 方法更加灵活。

第三，SASL，全称：Simple Authentication Security Layer, 用于安全认证，又分为 GSSAPI 、Kerberos、NTLM。

第四，JAAS，全称：Java Authentication and Authorization Service，Java 提供的安全认证服务，Kafka 使用 SASL 协议时，通过 JAAS 实现认证，需要配置 jaas 认证文件。

1、Kafka 服务配置文件

进入 kafka 应用的 config 目录，修改 server.properties 文件，在尾部添加认证协议配置：

# 修改listeners
listeners=SASL_PLAINTEXT://:9092

# 权限配置
allow.everyone.if.no.acl.found=true
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
super.users=User:admin

配置说明：

1. listeners=SASL_PLAINTEXT ，指定认证协议，默认无安全认证。
2. 检查 advertised.listeners 的协议类型也需要跟步骤 1 的协议类型一致。
3. ACL 权限未配置时，是否允许访问，默认为 false，如果用户读写一个 Topic，但是没有配置 ACL 权限，客户端会包认证失败错误。
4. security.inter.broker.protocol ，代理间通信所使用的安全协议。
5. sasl.mechanism.inter.broker.protocol，代理直接 SASL 安全认证协议类型。
6. sasl.enabled.mechanisms，Kafka 服务器的 SASL 安全认证协议类型 PLAIN。
7. super.users，超管用户名。

2、Kafka jaas 认证配置文件

在 config 目录下创建一个 jaas 配置文件 kafka-server-jaas-plain.conf ，使用简单帐号密码方式，配置内容如下：

KafkaServer {
    
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="admin"
  password="admin"
  user_admin="admin"
  user_test1="test1"
  user_test2="test2"
  user_test3="test3"
  user_test4="test4";
};

基本语法注意事项：

1. 第一行：固定名称指定 KafkaServer 端的配置。
2. 第二行：安全认证类名为 PlainLoginModule，与 Kafka 属性文件中的协议类型一致。此外，这个认证类也可以设置为 org.apache.kafka.common.security.scram.ScramLoginModule ，与 SCRAM 的认证类一致也能正常连接。
3. 第三、四行：服务端使用的帐号和密码。
4. 第五行，超管帐号的密码。
5. 后面都是预设普通帐号认证信息，语法为 user_真正的用户名=''密码"
6. 最后一行用户名密码定义后，必须以分号 ; 结尾。

3、Kafka 启动脚本添加认证文件路径的环境变量

Kafka 安全认证可以直接通过环境变量 -Djava.security.auth.login.config 设置，修改 Kafka 启动脚本 kafka-start-server.sh 文件最后一行，增加一个参数指向第二步的 jaas 配置文件的绝对路径：

exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/Downloads/kafka_2.11-2.3.1/config/kafka-server-jaas-plain.conf kafka.Kafka "$@"

4、启动 Kafka

第一步，启动 Zookeeper

bin/zookeeper-server-start.sh config/zookeeper.properties

第二步，启动 Kafka

bin/kafka-server-start.sh config/server.properties

5、Kafka 连接客户端认证配置

用户 Java 编写一个 Kafka 消费者，用帐号 test1/test1 进行安全认证，主要代码如下：

//默认是30000ms
kafkaProps.put(DataShareConstant.REQUEST_TIMEOUT, "5000");
kafkaProps.put("transaction.timeout.ms", "5000");
kafkaProps.put("max.block.ms", "5000"); // 该属性决定连接超时的
kafkaProps.setProperty("security.protocol", SecurityProtocol.SASL_PLAINTEXT.name);
kafkaProps.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
kafkaProps.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"test1\" password=\"test1\";");

注意，这里设置 JAAS 属性需要与 Kafka Server 中的属性一致：

1. 安全认证协议类型：security.protocol = SASL_PLAINTEXT
2. SASL 的认证机制 = PLAIN
3. SASL 认证配置信息，PlainLoginModule，帐号密码来自第二步配置预设的帐号。

值得注意的是，在做 Kafka 连通性测试时，因为 Kafka 存在失败重试机制，建议用同步请求，配置 max.block.ms 设置连接超时时间，迅速获取连接结果。

它默认是一分钟，如果需要通过页面配置 Kafka 信息，这个默认时间是不友好的。

启示录

Plain 模式的 SASL 安全认证实践起来比较容易，主要就是注意客户端和服务端协议类型的一致。

有几个常见问题：

1、如果服务器配置是 PLAIN，而客户端使用 SCRAM，则会报异常：

org.apache.kafka.common.errors.UnsupportedSaslMechanismException: Client SASL mechanism 'SCRAM-SHA-256' not enabled in the server, enabled mechanisms are [PLAIN]

2、客户端使用普通帐号，没有配置 ACL 权限时，默认写入 Kafka 时会报认证失败错误：Invalid username or password，这时帐号密码没问题的话，需要检查 ACL 权限。