使用 kafka-topics.sh 配置 Kafka 认证

1. 概述

Apache Kafka 是一个强大的平台，用于构建实时数据管道和流式应用。在生产环境中部署 Kafka 集群时，必须启用 Kafka 提供的安全机制来保护集群。要访问启用了认证的 Kafka 服务，客户端必须配置相应的凭证信息才能正常通信。

本文将介绍 Kafka 的认证机制，并演示如何配置 kafka-topics.sh 脚本与 Kafka 服务进行认证交互。

2. Kafka 的安全机制

Apache Kafka 提供了多种安全特性来增强集群安全性。其中，Kafka 服务端监听器可以配置支持的安全协议，从而控制客户端如何连接和认证。这些协议决定了客户端与服务端之间的认证方式以及是否加密传输数据。

2.1 安全协议

Kafka 支持四种安全协议：PLAINTEXT、SSL、SASL_PLAINTEXT 和 SASL_SSL。它们的主要区别在于是否启用认证和是否加密数据传输。

安全协议	是否启用认证	数据传输是否加密
PLAINTEXT	❌	❌
SSL	✅（通过 mTLS）	✅
SASL_PLAINTEXT	✅（通过 SASL）	❌
SASL_SSL	✅（通过 SASL）	✅

PLAINTEXT：不启用任何认证和加密，适合测试环境，生产环境禁用。
SSL：通过 SSL/TLS 建立加密通道，支持 mTLS（双向认证）。
SASL_PLAINTEXT：使用 SASL 框架进行认证，但不加密数据传输。
SASL_SSL：结合 SASL 认证和 SSL 加密，是最安全的协议。

2.2 SASL 认证机制

在使用 SASL_PLAINTEXT 或 SASL_SSL 协议时，必须指定具体的 SASL 认证机制。Kafka 支持多种 SASL 机制，如：

PLAIN：基于用户名和密码的简单认证
GSSAPI：Kerberos 认证
SCRAM-SHA-512：基于挑战-响应机制的认证
OAUTHBEARER：OAuth 2.0 令牌认证

**认证机制通常表示为 SASL/**，例如使用 PLAIN 机制时就是 SASL/PLAIN。

本文将以 SASL/PLAIN 为例进行演示，因为它配置简单，适合入门。

3. 使用 SASL/PLAIN 认证 Kafka 客户端

当客户端连接启用了 SASL 认证的 Kafka 服务时，需要响应服务器的认证挑战。在 SASL/PLAIN 情况下，客户端需提供用户名和密码进行身份验证。

Java 客户端通常使用 JAAS（Java Authentication and Authorization Service）处理 SASL 认证流程。Kafka 自带的脚本（如 kafka-topics.sh）也通过 JAAS 配置认证信息。

3.1 配置 JAAS 认证

我们可以通过 sasl.jaas.config 属性配置客户端的认证信息。示例如下：

$ cat client-config.properties
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="bob" password="bobpassword";
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

PlainLoginModule：用于 PLAIN 机制的认证模块
username 和 password：客户端用于认证的凭据
security.protocol：指定使用 SASL_PLAINTEXT 协议
sasl.mechanism：指定 SASL 认证机制为 PLAIN

3.2 使用配置文件运行脚本

在运行 Kafka 客户端脚本时，使用 --command-config 参数指定配置文件：

$ ./kafka-topics.sh --list --bootstrap-server localhost:9092 --command-config client-config.properties

脚本启动时会加载配置文件中的 JAAS 设置，并在连接 Kafka 服务端时进行认证。

4. 实战演示

我们将通过一个完整的例子来演示如何配置 Kafka 服务端并使用 kafka-topics.sh 进行认证连接。

4.1 部署启用了 SASL_PLAINTEXT 的 Kafka 服务

我们使用 Docker 部署一个启用了 SASL_PLAINTEXT 协议的 Kafka 实例：

# docker-compose.yaml
version: "3"
services:
  kafka:
    image: 'bitnami/kafka:latest'
    ports:
      - '9094:9094'
    environment:
      - KAFKA_ENABLE_KRAFT=yes
      - KAFKA_CFG_PROCESS_ROLES=broker,controller
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_LISTENERS=SECURE://:9094,CONTROLLER://:9093
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=SECURE:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT
      - KAFKA_CFG_ADVERTISED_LISTENERS=SECURE://127.0.0.1:9094
      - KAFKA_NODE_ID=1
      - KAFKA_CLIENT_USERS=bob
      - KAFKA_CLIENT_PASSWORDS=bobpassword
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@localhost:9093

KAFKA_CFG_LISTENERS：定义监听器
KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP：将监听器映射到对应的安全协议
KAFKA_CLIENT_USERS / KAFKA_CLIENT_PASSWORDS：设置客户端用户名和密码

启动服务：

$ docker-compose up -d

检查日志确认服务已启动：

$ docker logs -n 5 kafka-authentication-kafka-1
...
INFO [KafkaRaftServer nodeId=1] Kafka Server started

4.2 未配置认证时连接 Kafka

尝试在未配置认证的情况下连接 Kafka：

$ ./kafka-topics.sh --create --topic Messages --bootstrap-server localhost:9094
Error while executing topic command : Timed out waiting for a node assignment. Call: createTopics

连接失败，原因是客户端未通过认证。

4.3 配置认证后连接 Kafka

创建认证配置文件：

$ cat client-config.properties
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="bob" password="bobpassword"; 
security.protocol=SASL_PLAINTEXT 
sasl.mechanism=PLAIN

使用配置文件运行脚本：

$ ./kafka-topics.sh --create --topic Messages --bootstrap-server localhost:9094 --command-config client-config.properties
Created topic Messages.

$ ./kafka-topics.sh --list --bootstrap-server localhost:9094 --command-config client-config.properties
Messages

✅ 成功创建并列出 topic，说明认证配置生效。

Kafka sasl plaintext listener

5. 总结

本文介绍了 Kafka 的安全协议和 SASL 认证机制，并演示了如何配置 kafka-topics.sh 使用 SASL/PLAIN 认证方式连接启用了 SASL_PLAINTEXT 的 Kafka 服务。

核心要点总结如下：

✅ Kafka 支持 PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL 四种安全协议
✅ SASL_PLAINTEXT 是一种不加密但支持认证的协议
✅ 客户端通过 JAAS 配置 SASL 认证信息
✅ 使用 --command-config 参数指定配置文件
✅ 示例中使用 bitnami/kafka 镜像快速搭建启用了 SASL 的 Kafka 服务

📌 踩坑提醒：

配置认证信息时，注意 sasl.jaas.config 的格式必须严格正确，否则认证失败
Kafka 服务端的监听器和安全协议映射必须一致，否则客户端无法连接
生产环境建议使用 SASL_SSL 而非 SASL_PLAINTEXT，避免密码明文传输

通过本文的演示，你可以快速掌握 Kafka 客户端认证配置的基本方法，并将其应用到实际项目中。

Persistence

REST

Security