Spring Boot应用对接AWS MSK的正确消息消费配置咨询

本地Spring Boot应用使用以下配置可正常消费Kafka消息：

spring:
  cloud:
    stream:
      kafka:
        binder:
          replicationFactor: 1
          auto-create-topics: true
          brokers: localhost:9092
      bindings:
        binding-in-sse:
          destination: sse-topic
          content-type: text/plain
          group: earlywage
        binding-out-sse:
          destination: sse-topic
          content-type: text/plain
          group: earlywage

现需对接Dev环境的AWS MSK集群，MSK参数如下：

3个分区、3个副本、2个Broker、SASL/SCRAM认证、retention.ms=604800000、max.message.bytes=2097164

当前在同私有VPC内使用以下配置无法正常消费，请求给出正确的消息消费配置：

spring:
  cloud:
    stream:
      kafka:
        binder:
          replicationFactor: 1
          auto-create-topics: true
          brokers:
            - b-1.****.***.c2.kafka.REGION.amazonaws.com:PORT
            - b-2.****.***.c2.kafka.REGION.amazonaws.com:PORT
          configuration:
            security.protocol: SASL_PLAINTEXT
            sasl.mechanism: SCRAM-SHA-512
            sasl:
              jaas:
                config: org.apache.kafka.common.security.scram.ScramLoginModule required username="***" password="*****";
      bindings:
        binding-in-sse:
          destination: sse-topic
          content-type: text/plain
          group: earlywage
        binding-out-sse:
          destination: sse-topic
          content-type: text/plain
          group: earlywage

修正后的完整配置 #

spring:
  cloud:
    stream:
      kafka:
        binder:
          # 关闭自动创建topic，MSK集群的topic需提前按给定参数创建（3分区、3副本）
          auto-create-topics: false
          brokers:
            - b-1.****.***.c2.kafka.REGION.amazonaws.com:PORT
            - b-2.****.***.c2.kafka.REGION.amazonaws.com:PORT
          configuration:
            security.protocol: SASL_PLAINTEXT
            sasl.mechanism: SCRAM-SHA-512
            # YAML中用单引号包裹避免转义问题，替换为实际的MSK用户名和密码
            sasl.jaas.config: 'org.apache.kafka.common.security.scram.ScramLoginModule required username="your-msk-username" password="your-msk-password";'
            # 匹配MSK的消息大小限制
            max.message.bytes: 2097164
            fetch.max.bytes: 2097164
            # 可选：根据业务需求设置offset重置策略
            auto.offset.reset: earliest
      bindings:
        binding-in-sse:
          destination: sse-topic
          content-type: text/plain
          group: earlywage
          consumer:
            # 消费并发数匹配MSK分区数，提升消费能力
            concurrency: 3
            # 可选：开启批量消费（根据业务场景调整）
            batch-mode: true
        binding-out-sse:
          destination: sse-topic
          content-type: text/plain
          group: earlywage
          producer:
            # 匹配MSK的消息大小限制，避免发送失败
            max-message-size: 2097164

关键修正点 #

• 关闭自动创建Topic：MSK集群的Topic需提前按给定参数创建，本地配置的replicationFactor:1与MSK的3副本要求冲突，会导致自动创建失败，因此关闭auto-create-topics。
• JAAS配置转义处理：YAML中用单引号包裹JAAS配置内容，避免双引号转义问题，确保用户名和密码能被正确解析。
• 匹配消息大小限制：同步配置生产者和消费者的消息大小参数，与MSK的max.message.bytes保持一致，避免因大小限制导致消息发送或接收失败。
• 消费并发适配：设置concurrency:3与MSK的3个分区对应，保证每个分区都有对应的消费线程，充分利用集群资源。
• 安全配置验证：确认security.protocol和sasl.mechanism与MSK的SASL/SCRAM认证配置完全匹配，这是认证通过的核心前提。

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