请求协助配置Kafka、ZooKeeper、NiFi及Java Producer的SASL_PLAINTEXT认证

1. ZooKeeper 配置（zookeeper.properties） #

本地环境若Broker与ZooKeeper无需严格认证可简化配置，如需完整SASL链路，添加以下配置：

# 启用SASL认证
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000

同时创建zookeeper_jaas.conf文件（路径自行指定）：

Server {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="zkadmin"
    password="zkadmin-secret"
    user_zkadmin="zkadmin-secret";
};

启动ZooKeeper时指定JAAS配置：

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

2. Kafka Broker 配置（server.properties） #

配置Broker同时支持PLAINTEXT（兼容原有SpringBoot Producer）和SASL_PLAINTEXT（供NiFi使用）：

# 监听地址，同时开放两种协议端口
listeners=PLAINTEXT://localhost:9092,SASL_PLAINTEXT://localhost:9093
advertised.listeners=PLAINTEXT://localhost:9092,SASL_PLAINTEXT://localhost:9093

# 启用SASL认证
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN

# 配置Broker的SASL用户信息（直接写在server.properties里，无需额外JAAS文件）
listener.name.sasl_plaintext.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
    username="kafkaadmin" \
    password="kafkaadmin-secret" \
    user_nifiuser="nifi-secret" \
    user_springuser="spring-secret";

说明：这里创建了3个用户，kafkaadmin用于Broker内部通信，nifiuser给NiFi用，springuser给SpringBoot Producer（若要切换到SASL模式）。

3. SpringBoot Producer 配置 #

3.1 保持原有PLAINTEXT模式（无需修改） #

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

3.2 切换到SASL PLAIN模式（更安全） #

spring:
  kafka:
    bootstrap-servers: localhost:9093
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    properties:
      sasl.mechanism: PLAIN
      security.protocol: SASL_PLAINTEXT
      sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required \
          username="springuser" \
          password="spring-secret";

4. NiFi ConsumeKafka 2.0 处理器配置 #

打开ConsumeKafka_2_0处理器，设置以下核心参数：

• Bootstrap Servers: localhost:9093
• Security Protocol: SASL_PLAINTEXT
• SASL Mechanism: PLAIN
• SASL JAAS Configuration: org.apache.kafka.common.security.plain.PlainLoginModule required username="nifiuser" password="nifi-secret";
• Topic Names: 填写你的目标Kafka主题
• Group ID: 自定义消费组ID（比如nifi-kafka-consumer-group）

其他参数（如序列化方式、超时时间）保持默认即可，配置完成后启动处理器，就能正常消费Kafka消息。

内容的提问来源于stack exchange，提问作者Ram