Spring Boot启动创建Kafka Topic时配置异常问题排查

问题描述 #

使用Spring Boot 3.4.1 + Spring Kafka 3.3.1，启动时通过TopicBuilder创建Kafka Topic失败，因为它默认使用localhost:9092作为bootstrap地址，但KafkaTemplate能正确使用自定义的kafka-1.hello.com:9092, kafka-2.hello.com:9092配置发送消息。核心疑问：

• 为什么TopicBuilder和KafkaTemplate使用不同的Kafka配置？
• 如何复用已有的Producer配置给Topic创建逻辑？

原因分析 #

• KafkaTemplate依赖你自定义的ProducerFactory实例，因此会继承其中配置的bootstrap.servers参数。
• 启动时自动创建Topic的逻辑依赖Kafka AdminClient，Spring Kafka默认会自动初始化AdminClient，但如果未专门配置AdminClient的参数，它会使用默认值（如localhost:9092），且该配置与ProducerFactory的配置相互独立。
• TopicBuilder仅用于构建NewTopic对象，它的configs方法是设置Topic自身的属性（如cleanup.policy、retention.ms），而非AdminClient的连接配置。

解决方案 #

有两种简洁有效的解决方式：

方法1：全局配置AdminClient参数（推荐） #

直接为AdminClient指定bootstrap地址，可通过配置文件或Java配置类实现：

方式1.1：通过配置文件

在application.properties或application.yml中添加：

spring.kafka.admin.bootstrap-servers=${KAFKA_SERVERS}

方式1.2：通过Java配置类定义AdminClient配置

在你的KafkaProducerConfiguration中添加AdminClient配置Bean：

@Bean
public Map<String, Object> adminClientConfig() {
    Map<String, Object> configs = new HashMap<>();
    configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddresses);
    return configs;
}

Spring Kafka会自动使用该配置初始化AdminClient，用于启动时的Topic创建逻辑。

方法2：复用ProducerFactory配置（可选） #

如果你想复用已有的ProducerFactory配置，可将其转换为Map<String, String>后手动初始化AdminClient创建Topic，但这种方式不如方法1简洁，仅适用于特殊场景：

@Bean
public NewTopic topic() {
    // 转换ProducerFactory配置为字符串类型
    Map<String, String> adminConfigs = producerFactory().getConfigurationProperties()
            .entrySet().stream()
            .collect(Collectors.toMap(
                    Map.Entry::getKey,
                    entry -> entry.getValue().toString()
            ));
    // 手动创建AdminClient并创建Topic
    try (AdminClient adminClient = AdminClient.create(adminConfigs)) {
        CreateTopicsResult result = adminClient.createTopics(Collections.singleton(
                new NewTopic(KAFKA_TOPIC, 1, (short) 1)
        ));
        result.all().get(); // 等待创建完成
    } catch (InterruptedException | ExecutionException e) {
        log.error("Failed to create topic", e);
        throw new RuntimeException(e);
    }
    return TopicBuilder.name(KAFKA_TOPIC)
            .partitions(1)
            .replicas(1)
            .build();
}

修改后的完整配置类示例（方法1.2） #

@Slf4j
@Configuration
@RequiredArgsConstructor
public class KafkaProducerConfiguration {

    public static final String KAFKA_TOPIC = "topic1";

    @Value("${KAFKA_SERVERS}")
    private String bootstrapAddresses;

    @Bean
    public ProducerFactory<String, Event> producerFactory() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddresses);
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        return new DefaultKafkaProducerFactory<>(configs);
    }

    @Bean
    public KafkaTemplate<String, Event> kafkaTemplate() {
        var factory = producerFactory();
        log.debug("initializing a KafkaTemplate using the following setting: {{}}", configurationToString(factory));
        return new KafkaTemplate<>(factory);
    }

    // 添加AdminClient配置Bean
    @Bean
    public Map<String, Object> adminClientConfig() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddresses);
        return configs;
    }

    @Bean
    public NewTopic topic() {
        log.debug("creating a new topic kafka topic: {}", KAFKA_TOPIC);
        return TopicBuilder.name(KAFKA_TOPIC)
                .partitions(1)
                .replicas(1)
                .build();
    }

    private String configurationToString(ProducerFactory<String, Event> producerFactory) {
        var sb = new StringBuilder();
        producerFactory.getConfigurationProperties()
                .forEach((key, value) -> sb.append(String.format("\"%s\": \"%s\"", key, value)));
        return sb.toString();
    }
}

关键注意点 #

• 不要混淆TopicBuilder的configs方法用途：它仅用于设置Topic自身的属性，而非AdminClient的连接参数。
• AdminClient的配置与ProducerFactory是独立的，必须单独指定才能让Topic创建逻辑使用正确的Kafka集群地址。

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