Kafka Streams：如何通过Java自动创建源主题？是否建议此操作？

问题背景 #

我正在开发Kafka Streams应用，流程为：消息推入入站主题（源主题）→ 经Streams处理后推至出站主题 → 我的Java应用从出站主题消费消息。首次运行时触发报错：

The following source topics are missing/unknown: [local.inbound.topic]. Please make sure all source topics have been pre-created before starting the Streams application.

已知Kafka Streams要求源主题必须提前创建，现有两个备选方案：请管理员部署前创建主题，或通过代码自动创建。想明确两个问题：能否通过Java在Streams运行器启动前自动创建入站主题？该操作是否合理？

解答 #

1. 可以通过Java自动创建入站主题 #

借助Kafka官方提供的AdminClient API，就能在启动Kafka Streams应用前完成主题的检查与创建。以下是极简实现示例：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class KafkaTopicInitializer {
    public static void ensureTopicExists(String bootstrapServers, String topicName, int partitions, short replicationFactor) throws ExecutionException, InterruptedException {
        Properties adminProps = new Properties();
        adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        
        try (AdminClient adminClient = AdminClient.create(adminProps)) {
            // 先检查主题是否已存在
            if (!adminClient.listTopics().names().get().contains(topicName)) {
                NewTopic targetTopic = new NewTopic(topicName, partitions, replicationFactor);
                adminClient.createTopics(Collections.singleton(targetTopic)).all().get();
                System.out.printf("主题 %s 已成功创建%n", topicName);
            }
        }
    }
}

在调用KafkaStreams.start()启动流应用之前，直接调用ensureTopicExists方法即可确保源主题存在。

2. 操作合理性需分场景判断 #

适合自动创建的场景

• 开发/测试环境：无需手动介入创建主题，能快速搭建测试环境，提升开发迭代效率。
• 本地单实例Kafka：无复杂权限管控，自动创建不会引入额外风险，适合个人调试使用。

不推荐自动创建的场景

• 生产环境：生产集群通常有严格的权限体系与主题管理规范，自动创建可能违反运维流程；且代码中硬编码的分区数、副本因子等配置可能不符合生产标准，引发潜在问题。
• 多团队协作场景：主题属于共享资源，自动创建易导致命名冲突、配置不一致等问题，应由统一的运维团队或标准化流程管理主题生命周期。

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