Spring Cloud Stream消费Kafka主题时无法获取分区信息问题

问题背景 #

生产消息正常，通过kafkacat可确认消息已写入主题segnnel，但消费端启动时偶尔抛出以下错误，自动重试后问题依旧：

Caused by: java.lang.RuntimeException: Failed to obtain partition information for the topic segnnel
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.lambda$getPartitionsForTopic$9(KafkaTopicProvisioner.java:658)

o.s.cloud.stream.binding.BindingService - Failed to create consumer binding; retrying in 30 seconds
org.springframework.cloud.stream.binder.BinderException: Cannot initialize binder checking the topic (segnnel):
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.getPartitionsForTopic(KafkaTopicProvisioner.java:685)

使用原生@KafkaListener可正常消费，但业务要求必须使用Spring Cloud Stream。

核心解决方案 #

1. 清理冗余配置，强化元数据重试逻辑

原配置存在Kafka连接参数重复、元数据重试不足的问题，调整后配置如下：

spring:
  cloud:
    function:
      definition: chents
    stream:
      bindings:
        chents-in-0:
          destination: segnnel 
          group: console-consumer-20965  
          consumer:
            max-attempts: 2
        chents-out-0:
          destination: segnnel   
      kafka:
        binder:
          brokers: localhost:19092
          # 增加元数据获取的重试与过期配置
          configuration:
            metadata:
              max:
                age: 30000
              retry:
                max: 5
                backoff:
                  ms: 1000
        bindings:
          chents-in-0:
            consumer:
              enableDlq: false
              ack-mode: record
              key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
              value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
              properties:
                spring.json.trusted.packages: "*"

• 移除重复的bootstrap-servers，统一使用binder.brokers配置连接地址
• 添加metadata.retry参数，增加元数据获取的重试次数与退避时间，解决启动时Kafka元数据未就绪的问题

2. 调整绑定重试策略

默认绑定重试间隔30秒，可缩短间隔并增加总重试次数，避免长时间等待：

spring:
  cloud:
    stream:
      binding-service:
        retry-interval: 5000
        max-retries: 10

3. 优化消费者异步处理逻辑

原消费者使用Mono但未确保操作完成，可能导致上下文提前关闭，调整代码如下：

@Configuration 
public class ChannelSubscriber {

    private final Repository repository;
    private static final Mapper mapper;

    public ChannelSubscriber(Repository repository) {
        this.repository = repository;
    }

    @Bean
    Consumer<EventData<Channel>> chents() {
        return e -> {
            try {
                Mono.just(e)
                    .map(mapper::mapToEntity)
                    .flatMap(repository::save)
                    .doOnError(ex -> log.error("Error processing event: ", ex))
                    .block(); // 确保异步操作执行完成
            } catch (Exception ex) {
                log.error("Failed to process event", ex);
                // 抛出异常触发重试（配合max-attempts配置）
                throw new RuntimeException(ex);
            }
        };
    }

}

• 添加block()确保异步数据库操作完成，避免Spring Cloud Stream误判消费状态
• 增加全局异常捕获，确保错误能被正确记录并触发重试机制

4. 验证Kafka集群就绪状态

若使用本地Docker或测试环境，需确保Kafka Broker在服务启动前已完全初始化，可在启动脚本中添加健康检查逻辑，避免服务在Kafka未就绪时启动。

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