Spring Boot集成Kafka：如何通过Java设置Topic配置？

项目基础配置 #

pom.xml 核心依赖 #

...
<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>3.2.5</version>
    <relativePath />
</parent>
....
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
...

Kafka监听器代码 #

@KafkaListener(topics = "X_OUT", autoStartup = "true")
public void listenTo(ConsumerRecord<String, String> cr, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
  // 业务逻辑处理
}

当前程序启动后会自动创建X_OUT Topic，Kafka功能运行正常。现在需要修改该Topic的两项配置：

• 将min.insync.replicas设置为1
• 将副本因子（ReplicationFactor）设置为4

目前找到的控制台工具修改方式：

# 修改min.insync.replicas
kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type topics --entity-name X_OUT --add-config min.insync.replicas=1

# 修改副本因子（需配合JSON配置文件）
kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file reassignment.json --execute

问题：不想使用控制台工具，能否在Spring Boot应用中通过Java代码自动完成这些配置修改（创建Topic时设置或后期动态修改）？

解决方案 #

完全可以通过Java代码实现，分两种场景处理：

1. 创建Topic时直接指定目标配置 #

利用Spring Kafka提供的KafkaAdmin组件，在应用启动时自动创建符合配置要求的Topic，无需手动干预。

配置示例

import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaAdmin;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaTopicConfig {

    @Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        return new KafkaAdmin(configs);
    }

    @Bean
    public NewTopic xOutTopic() {
        // 配置Topic参数：名称、分区数、副本因子，以及自定义配置
        Map<String, String> topicConfigs = new HashMap<>();
        topicConfigs.put("min.insync.replicas", "1");

        return new NewTopic("X_OUT", 1, (short) 4) // 这里的4就是ReplicationFactor
                .configs(topicConfigs);
    }
}

注意：如果Topic已存在，KafkaAdmin默认不会覆盖现有配置；若需强制更新新增配置项，可设置KafkaAdmin的initializeTopics属性为true，但该操作无法修改已存在的副本因子（副本因子属于Topic元数据，并非动态配置项）。

2. 动态修改已存在的Topic配置 #

对于已创建的Topic，需分两类配置分别处理：

2.1 修改min.insync.replicas（动态配置项）

使用Kafka的AdminClient调用alterConfigs方法直接修改：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.AlterConfigsResult;
import org.apache.kafka.common.config.ConfigResource;
import org.springframework.stereotype.Component;

import jakarta.annotation.PostConstruct;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;

@Component
public class KafkaTopicConfigUpdater {

    @PostConstruct
    public void updateMinInsyncReplicas() throws ExecutionException, InterruptedException {
        Map<String, Object> adminConfigs = new HashMap<>();
        adminConfigs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        try (AdminClient adminClient = AdminClient.create(adminConfigs)) {
            ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, "X_OUT");
            Map<String, String> configs = new HashMap<>();
            configs.put("min.insync.replicas", "1");

            AlterConfigsResult result = adminClient.alterConfigs(Collections.singletonMap(resource, configs));
            result.all().get(); // 等待修改完成
        }
    }
}

2.2 修改副本因子（ReplicationFactor）

副本因子属于Topic分区元数据，无法直接修改，需通过重新分配分区副本的方式实现，对应控制台kafka-reassign-partitions.sh的功能：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.AlterPartitionReassignmentsResult;
import org.apache.kafka.clients.admin.ReassignablePartition;
import org.apache.kafka.common.Node;
import org.springframework.stereotype.Component;

import jakarta.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;

@Component
public class KafkaReplicationFactorUpdater {

    @PostConstruct
    public void updateReplicationFactor() throws ExecutionException, InterruptedException {
        Map<String, Object> adminConfigs = new HashMap<>();
        adminConfigs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        try (AdminClient adminClient = AdminClient.create(adminConfigs)) {
            // 假设Kafka集群有4个broker，节点ID分别为0、1、2、3
            List<Node> newReplicas = List.of(
                    new Node(0, "localhost", 9092),
                    new Node(1, "localhost", 9093),
                    new Node(2, "localhost", 9094),
                    new Node(3, "localhost", 9095)
            );

            // 为X_OUT的所有分区分配新副本，这里假设只有1个分区（ID为0）
            // 实际场景可通过adminClient.describeTopics获取所有分区信息
            ReassignablePartition partition = new ReassignablePartition("X_OUT", 0, newReplicas);
            Map<String, List<ReassignablePartition>> reassignment = new HashMap<>();
            reassignment.put("X_OUT", List.of(partition));

            AlterPartitionReassignmentsResult result = adminClient.alterPartitionReassignments(reassignment);
            result.all().get(); // 等待重新分配完成
        }
    }
}

注意：重新分配副本会触发Kafka的副本同步，占用集群资源，建议在低峰期执行；同时需确保集群中有足够的broker节点（至少4个，对应副本因子4）。

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