kafka配置SASL_PLAINTEXT简单认证-EW帮帮网

Kafka + ZooKeeper 开启 SASL_PLAINTEXT 认证（PLAIN机制）最全实战教程

💡 本教程将手把手教你如何为 Kafka 配置基于 SASL_PLAINTEXT + PLAIN 的用户名密码认证机制，包含 Kafka 与 ZooKeeper 的全部配置，适合入门。

🎯 教程目标

Kafka 客户端连接 Kafka Broker 时需要用户名密码验证；
Kafka 与 ZooKeeper 之间通信也启用 SASL 认证；
使用 PLAIN 机制，无需 TLS/SSL 证书（比 SASL_SSL 简单）；
可用于本地开发环境或非安全生产环境。

🛠 环境准备

Kafka：3.6.0
ZooKeeper：3.6+
Java 8+
操作系统：Linux / WSL / Mac / Windows（推荐用 WSL）
镜像：
zookeeper bitnami_zookeeper:3.8.4
kafka bitnami/kafka:3.6.0

1️⃣ 配置 ZooKeeper 认证（Server端）

配置环境变量

env:
  - name: TZ
     value: Asia/Shanghai
   - name: ALLOW_ANONYMOUS_LOGIN
     value: 'no'
   - name: JVMFLAGS
     value: '-Xmx1g'
   - name: ZOO_ENABLE_AUTH
     value: 'yes'
   - name: ZOO_SERVER_USERS
     value: kafka
   - name: ZOO_SERVER_PASSWORDS
     value: zookeeper@2025
   - name: ZOO_SERVERS
     value: 'zk-cluster-auth-0.zk-cluster-auth-headless:2888:3888,zk-cluster-auth-1.zk-cluster-auth-headless:2888:3888,zk-cluster-auth-2.zk-cluster-auth-headless:2888:3888'

配置项	说明
`TZ=Asia/Shanghai`	设置容器的时区为中国标准时间（CST/UTC+8），方便日志与系统时间保持一致。
`ALLOW_ANONYMOUS_LOGIN=no`	禁用匿名连接 ZooKeeper，必须通过用户名密码认证。建议生产环境使用。
`JVMFLAGS='-Xmx1g'`	配置 ZooKeeper JVM 最大内存为 1GB，防止 OOM（默认可能太小）。
`ZOO_ENABLE_AUTH=yes`	启用 ZooKeeper 身份认证（基于 SASL 的认证机制，如 PLAIN）。必须配合下面的用户密码使用。
`ZOO_SERVER_USERS=kafka`	设置 ZooKeeper 允许的用户名，多个用户用逗号分隔。这里是 `kafka`。
`ZOO_SERVER_PASSWORDS=zookeeper@2025`	对应上面的用户的密码。如果多个用户，用逗号一一对应写。

服务配置

kind: Service
apiVersion: v1
metadata:
  name: zk-cluster-auth-headless
  namespace: zhubayi-common
  labels:
    app: kafka-cluster-auth
  annotations:
    kubesphere.io/alias-name: zk-cluster-auth
    kubesphere.io/creator: admin
    kubesphere.io/serviceType: statefulservice
spec:
  ports:
    - name: client
      protocol: TCP
      port: 2181
      targetPort: 2181
    - name: server
      protocol: TCP
      port: 2888
      targetPort: 2888
    - name: leader-select
      protocol: TCP
      port: 3888
      targetPort: 3888
    - name: adminserver
      protocol: TCP
      port: 8080
      targetPort: 8080
  selector:
    app: zk-cluster-auth
  clusterIP: None
  clusterIPs:
    - None
  type: ClusterIP
  sessionAffinity: None
  ipFamilies:
    - IPv4
  ipFamilyPolicy: SingleStack
  internalTrafficPolicy: Cluster

启动脚本

 command:
      - sh
      - '-c'
      - >
        export ZOO_SERVER_ID=$((${HOSTNAME##*-}+1))

        exec /opt/bitnami/scripts/zookeeper/entrypoint.sh
        /opt/bitnami/scripts/zookeeper/run.sh

2️⃣ 配置 Kafka（Broker端）

🧩 设置环境变量

env:
  #  设置容器的时区
  - name: TZ
    value: Asia/Shanghai

  # 🔗 Kafka 连接的 ZooKeeper 地址（推荐使用 headless 服务）
  - name: KAFKA_CFG_ZOOKEEPER_CONNECT
    value: 'zk-cluster-auth-headless:2181'

  # Kafka 内存配置（JVM堆大小）
  - name: KAFKA_HEAP_OPTS
    value: '-Xmx2g'

  # offsets topic 的副本数（建议 >=3）
  - name: KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR
    value: '3'

  # 每个 topic 的默认分区数
  - name: KAFKA_CFG_NUM_PARTITIONS
    value: '5'

  # Kafka 日志保留时间（单位小时）
  - name: KAFKA_CFG_LOG_RETENTION_HOURS
    value: '72'  # 3天

  # Kafka 日志切分时间（单位小时）
  - name: KAFKA_CFG_LOG_ROLL_HOURS
    value: '72'

  # 单个 segment 的最大大小（1GB）
  - name: KAFKA_CFG_LOG_SEGMENT_BYTES
    value: '1073741824'

  # 是否允许自动创建 topic（开发环境可开启）
  - name: KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE
    value: 'true'

  # 允许删除 topic
  - name: KAFKA_CFG_DELETE_TOPIC_ENABLE
    value: 'true'

  # 是否允许自动进行 leader 重新平衡（建议关闭以避免扰动）
  - name: KAFKA_CFG_AUTO_LEADER_REBALANCE_ENABLE
    value: 'false'

  # Kafka 清理策略（delete / compact）
  - name: KAFKA_CFG_LOG_CLEANUP_POLICY
    value: delete

  # 单条消息最大大小（5MB）
  - name: KAFKA_CFG_MESSAGE_MAX_BYTES
    value: '5242880'

  # 客户端请求最大大小（4MB）
  - name: KAFKA_CFG_MAX_REQUEST_SIZE
    value: '4194304'

  # 批处理大小（生产者用）
  - name: KAFKA_CFG_BATCH_SIZE
    value: '16384'

  # 允许使用 PLAINTEXT 明文监听器（非加密，仅适合内网测试）
  - name: ALLOW_PLAINTEXT_LISTENER
    value: 'true'

  # broker 间通信监听器名称（INSIDE 表示内网）
  - name: KAFKA_INTER_BROKER_LISTENER_NAME
    value: INSIDE

  # 监听器与协议映射（INSIDE 和 OUTSIDE 都启用 SASL_PLAINTEXT）
  - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
    value: 'INSIDE:SASL_PLAINTEXT,OUTSIDE:SASL_PLAINTEXT'

  # 是否允许客户端连接时自动创建 topic
  - name: KAFKA_AUTO_CREATE_TOPICS_ENABLE
    value: 'true'

  # 与 SSL 无关，可忽略（如果未使用 TLS）
  - name: KAFKA_SSL_CLIENT_AUTH
    value: required

  # Kafka 控制器使用的认证机制（SASL PLAIN）
  - name: KAFKA_CFG_SASL_MECHANISM_CONTROLLER_PROTOCOL
    value: PLAIN

  # broker 间通信使用的认证机制（SASL PLAIN）
  - name: KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL
    value: PLAIN

  # SSL 类型（使用 Java KeyStore 格式）（此配置无效，可忽略，未启用 SSL）
  - name: KAFKA_TLS_TYPE
    value: JKS

  # 指定 broker 之间用哪个监听器通信（重复定义，建议只保留一次）
  - name: KAFKA_CFG_INTER_BROKER_LISTENER_NAME
    value: SASL_PLAINTEXT

  # 客户端连接使用的监听器（与上面保持一致）
  - name: KAFKA_CLIENT_LISTENER_NAME
    value: SASL_PLAINTEXT

  # Controller 用户名（用于 Kafka 内部通信）
  - name: KAFKA_CONTROLLER_USER
    value: kafka

  # Controller 用户密码
  - name: KAFKA_CONTROLLER_PASSWORD
    value: 123456

  # Broker 间通信用户名
  - name: KAFKA_INTER_BROKER_USER
    value: kafka

  # Broker 间通信密码
  - name: KAFKA_INTER_BROKER_PASSWORD
    value: 123456

  # 允许的客户端用户名（支持多个，逗号分隔）
  - name: KAFKA_CLIENT_USERS
    value: kafka

  # 客户端密码（顺序与用户一致）
  - name: KAFKA_CLIENT_PASSWORDS
    value: 123456

  # 与 ZooKeeper 通信时使用的认证协议（SASL 必须设置）
  - name: KAFKA_ZOOKEEPER_PROTOCOL
    value: SASL

  # Kafka 访问 ZooKeeper 时使用的用户名
  - name: KAFKA_ZOOKEEPER_USER
    value: kafka

  # Kafka 访问 ZooKeeper 时使用的密码（需与 ZooKeeper 中配置一致）
  - name: KAFKA_ZOOKEEPER_PASSWORD
    value: zookeeper@2025

启动脚本

 command:
   - sh
   - '-c'
   - >

     POD_NAME=$(hostname)

     echo "POD_NAME:$POD_NAME"

     REPLICA_INDEX=$(echo $POD_NAME | sed 's/.*-\([0-9]\)$/\1/')

     echo "REPLICA_INDEX:$REPLICA_INDEX"

     export KAFKA_CFG_NODE_ID=${POD_NAME##*-}

     export KAFKA_NODE_ID="$REPLICA_INDEX"


     PORT=$((REPLICA_INDEX + 30900)) PORT2=$((REPLICA_INDEX + 9093))


     export
     KAFKA_CFG_ADVERTISED_LISTENERS="INSIDE://:9092,OUTSIDE://192.168.1.5:$PORT"


     export KAFKA_CFG_LISTENERS="INSIDE://:9092,OUTSIDE://:$PORT2"

     exec /opt/bitnami/scripts/kafka/entrypoint.sh
     /opt/bitnami/scripts/kafka/run.sh

服务端口配置

spec:
  ports:
    - name: tcp-9092
      protocol: TCP
      port: 9093
      targetPort: 9093
      nodePort: 30900
    - name: tcp-9093
      protocol: TCP
      port: 9094
      targetPort: 9094
      nodePort: 30901
    - name: tcp-9094
      protocol: TCP
      port: 9095
      targetPort: 9095
      nodePort: 30902

3️⃣ Kafka 客户端配置（示例）

配置文件

spring:
  kafka:
    bootstrap-servers: 192.168.1.5:30900  # Kafka 集群地址（多个可逗号分隔）

    listener:
      ack-mode: MANUAL_IMMEDIATE  # 手动提交 offset（立即确认）

    consumer:
      custom-environment: dev # 自定义字段，可用于多环境区分，无实际作用
      auto-offset-reset: latest  # 无 offset 时从最新消息开始消费（避免重复）
      enable-auto-commit: false  # 禁用自动提交，改为手动提交 offset
      # auto-commit-interval: 2000  # 如果启用自动提交，间隔为 2 秒（此处已注释）
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer  # key 反序列化方式
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer  # value 反序列化方式
      max-poll-records: 50  # 每次 poll 最多拉取 50 条记录
      max-poll-interval-ms: 600000  # poll 最大间隔 10 分钟

    producer:
      retries: 0  # 不重试（生产失败直接报错）
      batch-size: 16384  # 批量发送最大字节数（默认 16KB）
      buffer-memory: 33554432  # 发送缓冲区大小（默认 32MB）
      key-serializer: org.apache.kafka.common.serialization.StringSerializer  # key 序列化
      value-serializer: org.apache.kafka.common.serialization.StringSerializer  # value 序列化

    ssl:
      trust-store-location:    # ⚠ 这里为空，因使用 SASL_PLAINTEXT（非 SSL），可留空或移除
      trust-store-password:

    properties:
      sasl:
        mechanism: PLAIN  # 使用 SASL PLAIN 机制（用户名密码）
        jaas:
          config: >-
            org.apache.kafka.common.security.scram.ScramLoginModule required
            username="kafka"
            password="123456";
          # JAAS 配置：用户名密码认证（注意 password 后有分号）

      security:
        protocol: SASL_PLAINTEXT  # 使用 SASL_PLAINTEXT 传输协议（非加密）

      ssl:
        endpoint:
          identification:
            algorithm: ""  # 空表示跳过主机名校验（SSL 时才生效，非必须）

🚨 注意事项

配置项	注意点
`SASL_PLAINTEXT`	明文传输用户名密码，不建议用于公网
`ScramLoginModule`	表示服务端配置的是 SCRAM（非 PLAIN）时使用，若是 PLAIN，应为 `PlainLoginModule`
`trust-store-location`	可删除或忽略（仅用于 `SASL_SSL` 或 `SSL` 场景）
`bootstrap-servers`	建议配置多个 broker IP，提升可靠性
`ack-mode: MANUAL_IMMEDIATE`	消费者业务失败时不会提交 offset，可重复消费

生产者配置

@Configuration
public class KafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    @Value("${spring.kafka.consumer.auto-offset-reset}")
    private String autoOffsetReset;
    @Value("${spring.kafka.consumer.enable-auto-commit}")
    private String enableAutoCommit;
    @Value("${spring.kafka.consumer.max-poll-records}")
    private String maxPollRecords;

    @Value("${spring.kafka.consumer.max-poll-interval-ms}")
    private String maxPollIntervalMs;

    @Value("${spring.profiles.active}")
    private String activeProfile;

    @Value("${spring.kafka.properties.security.protocol:SASL_SSL}")
    private String securityProtocol;
    @Value("${spring.kafka.properties.sasl.mechanism:PLAIN}")
    private String salsMechanism;
    @Value("${spring.kafka.properties.ssl.endpoint.identification.algorithm: \"\"}")
    private String identificationAlgorithm;
    @Value("${spring.kafka.properties.sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username=\"kafka\" password=\"123456\";}")
    private String jaasConfig;
    @Value("${spring.kafka.ssl.trust-store-password: \"\"}")
    private String trustStorePassword;
    @Value("${spring.kafka.ssl.trust-store-location: \"\"}")
    private String trustStoreLocation;



    @Bean
    public ConsumerFactory<String, String> daConsumerConfigFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, KafkaConsumerComponent.DAGROUP);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxPollIntervalMs);
        props.put("security.protocol", securityProtocol);
        props.put("sasl.mechanism", salsMechanism);
        props.put("ssl.endpoint.identification.algorithm", identificationAlgorithm);
        props.put("consumer.ssl.endpoint.identification.algorithm", identificationAlgorithm);
        props.put("sasl.jaas.config", jaasConfig);
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> daConsumerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(daConsumerConfigFactory());
        factory.setBatchListener(true); // 启用批量消费
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }
}

🚨 常见问题排查

错误	可能原因
`No principal`	未正确配置 JAAS 文件或 KAFKA_OPTS 未设置
`SaslAuthenticationException`	用户名密码不匹配
`Connection refused`	listener 或端口未正确绑定
`Cluster ID 不匹配`	ZooKeeper 路径配置错误或 meta.properties 被复用

✅ 总结

项目	配置方式
ZooKeeper 启用鉴权	`zookeeper_jaas.conf` + `zoo.cfg`
Kafka 启用鉴权	`kafka_server_jaas.conf` + `server.properties`
客户端连接 Kafka	`kafka_client_jaas.conf`

📎 参考资料

官方文档：https://kafka.apache.org/documentation/
Kafka 安全机制：https://kafka.apache.org/documentation/#security_sasl

kafka配置SASL_PLAINTEXT简单认证

Kafka + ZooKeeper 开启 SASL_PLAINTEXT 认证（PLAIN机制）最全实战教程

🎯 教程目标

🛠 环境准备

1️⃣ 配置 ZooKeeper 认证（Server端）

2️⃣ 配置 Kafka（Broker端）

🧩 设置环境变量

启动脚本

3️⃣ Kafka 客户端配置（示例）

🚨 常见问题排查

✅ 总结

📎 参考资料

网站公告

今日签到

热门文章

最新发布