Spring Cloud Stream连接Kafka获取Topic信息超时问题求助

问题背景 #

内部部署的Kafka服务与普通Spring Boot应用配合正常，但使用Spring Cloud Stream实现测试方案时，在获取Topic列表/预配Topic环节出现超时。已确认Kafka基础连接正常，仅该环节失败，关键报错为ProvisioningException，根因是TimeoutException。

相关配置与代码 #

build.gradle #

plugins {
    id 'java'
    id 'org.springframework.boot' version '3.1.3'
    id 'io.spring.dependency-management' version '1.1.3'
}

group = 'com'
version = '0.0.1-SNAPSHOT'

java {
    sourceCompatibility = '17'
}

repositories {
    mavenCentral()
}
ext {
    set('springCloudVersion', "2022.0.4")
}

dependencies {
    implementation 'org.springframework.cloud:spring-cloud-stream'
    implementation 'org.springframework.cloud:spring-cloud-starter-stream-kafka'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
    testImplementation 'org.springframework.cloud:spring-cloud-stream-test-binder'
}

dependencyManagement {
    imports {
        mavenBom "org.springframework.cloud:spring-cloud-dependencies:\${springCloudVersion}"
    }
}

tasks.named('test') {
    useJUnitPlatform()
}

application.yml #

spring:
  cloud:
    function:
      definition: doLightMeasured;onLightMeasured
    stream:
      bindings:
        doLightMeasured-out-0:
          destination: light_measured
        onLightMeasured-in-0:
          destination: light_measured
      kafka:
        binder:
          brokers: 10.72.88.234:30092

logging:
  level:
    root: debug
    org:
      springframework: debug

AsyncApiTestApplication.java #

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

import java.util.function.Consumer;
import java.util.function.Supplier;

@SpringBootApplication
public class AsyncApiTestApplication
{
private static final Logger logger = LoggerFactory.getLogger(AsyncApiTestApplication.class);

    public static void main( String[] args )
    {
        SpringApplication.run(AsyncApiTestApplication.class, args);
    }
    
    @Bean
    public Supplier<LightMeasured> doLightMeasured()
    {
        return () -> {
            // Add business logic here.
            return new LightMeasured();
        };
    }
    
    @Bean
    public Consumer<LightMeasured> onLightMeasured()
    {
        return data -> {
            // Add business logic here.
            logger.info(data.toString());
        };
    }

}

关键错误日志 #

2023-09-15T18:16:08.213+02:00 ERROR 13664 --- [           main] o.s.cloud.stream.binding.BindingService  : 创建生产者绑定失败；30秒后重试

org.springframework.cloud.stream.provisioning.ProvisioningException: 为light_measured执行资源预配时遇到异常
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.createTopic(KafkaTopicProvisioner.java:372) ~[spring-cloud-stream-binder-kafka-core-4.0.4.jar:4.0.4]
Caused by: java.util.concurrent.TimeoutException: null
    at java.base/java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1960) ~[na:na]

排查与解决步骤 #

1. 延长Kafka binder超时配置 #

Spring Cloud Stream Kafka binder默认超时可能无法适配内部Kafka的网络延迟，可通过以下配置延长超时时间：

spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: 10.72.88.234:30092
          # 延长Topic创建/查询超时，单位毫秒
          create-topic-timeout: 60000
          admin:
            properties:
              request.timeout.ms: 60000
              default.api.timeout.ms: 60000

2. 检查Kafka ACL权限 #

确认应用使用的Kafka账号具备Topic查询/创建权限。普通Spring Boot应用可能仅需生产/消费权限，但Spring Cloud Stream启动时会预检查或自动创建Topic，需额外权限：

• 若Kafka开启ACL，为账号添加Describe、Create权限（针对目标Topic或全局）
• 用Kafka命令行验证权限：

  kafka-acls.sh --bootstrap-server 10.72.88.234:30092 --list --principal User:your-app-user
  

3. 禁用自动Topic创建 #

若内部Kafka不允许客户端自动创建Topic，且light_measuredTopic已存在，可禁用自动创建逻辑避免预配检查：

spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: 10.72.88.234:30092
          auto-create-topics: false

4. 检查Kafka Broker网络配置 #

基础连接正常不代表元数据查询路径通畅：

• 确认Broker的listeners配置包含外部可访问的IP/hostname，避免使用内部专属地址
• 检查防火墙是否允许客户端访问Kafka的控制平面端口（你的环境是30092）

5. 验证版本兼容性 #

Spring Cloud 2022.0.4对应的Kafka binder适配Kafka 2.8.x至3.3.x，若内部Kafka版本超出此范围，需调整Spring Cloud版本或添加兼容配置

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