KafkaContainer中Topic无法自动创建的问题排查与解决咨询

问题背景 #

尝试通过KafkaContainer的withEnv("KAFKA_CREATE_TOPICS", "subs-topic:1:1")参数自动创建Topic失败，同时使用AdminClient的createTopics方法创建Topic也无效。相关配置代码如下：

ContainersConfiguration类 #

@Configuration
public class ContainersConfiguration {
    public static final DockerImageName KAFKA_IMAGE_NAME = DockerImageName.parse("confluentinc/cp-kafka:5.2.1");
    private final Network network = Network.newNetwork();

    @Bean(initMethod = "start", destroyMethod = "stop")
    public ZookeeperContainer zookeeperContainer(ZookeeperContainerConfiguration zookeeperContainerConfiguration) {
        return new ZookeeperContainer()
                .withNetwork(network)
                .withNetworkAliases(zookeeperContainerConfiguration.getHost());
    }


    @Bean(initMethod = "start", destroyMethod = "stop")
    public KafkaContainer kafkaContainer(KafkaContainerConfiguration kafkaContainerConfiguration,
                                         ZookeeperContainer zookeeperContainer) {
        return new KafkaContainer(KAFKA_IMAGE_NAME)
                .withNetwork(network)
                .withEnv("KAFKA_CREATE_TOPICS", "subs-topic:1:1")
                .withNetworkAliases(kafkaContainerConfiguration.getBootstrapHost())
                .withStartupTimeout(Duration.ofSeconds(180))
                .withExternalZookeeper(zookeeperContainer.getNetworkAliases().get(0) + ":" + CLIENT_PORT);
    }
}

KafkaContainerConfiguration类 #

@Configuration
@PropertySource(value = "classpath:services.properties")
@Getter
@Setter
public class KafkaContainerConfiguration {

    @NotNull
    @Value("${kafka.bootstrap.server.host}")
    private String bootstrapHost;

    @NotNull
    @Value("${kafka.data.topic}")
    private String dataTopic;

    @NotNull
    @Value("${kafka.notification.topic}")
    private String notificationTopic;

}

services.properties配置 #

kafka.data.topic=subs-topic
kafka.notification.topic=subs-topic-notifications

KafkaProducerConfiguration类 #

@Configuration
public class KafkaProducerConfiguration {

    @Bean(destroyMethod = "close")
    public KafkaProducer<String, String> kafkaProducer(Properties kafkaProperties) {
        return new KafkaProducer<>(kafkaProperties);
    }

    @Bean(destroyMethod = "close")
    public AdminClient kafkaAdminClient(Properties kafkaProperties) {
        return AdminClient.create(kafkaProperties);
    }

    @SneakyThrows
    @Bean
    public Properties kafkaProperties(KafkaContainer kafkaContainer) {
        Properties properties = new Properties();
        properties.load(new FileReader(new File(ClassLoader.getSystemResource("kafka.producer.properties").getFile())));
        properties.put("bootstrap.servers", kafkaContainer.getBootstrapServers());
        return properties;
    }

}

失败原因分析 #

1. KAFKA_CREATE_TOPICS环境变量无效：

   • confluentinc/cp-kafka:5.2.1镜像的自动创建Topic逻辑依赖容器启动时的初始化脚本，但Testcontainers的KafkaContainer封装可能覆盖了该脚本执行逻辑，导致手动设置的环境变量不生效。
   • 若Kafka容器未正确连接Zookeeper（比如CLIENT_PORT值错误），初始化脚本执行时会失败，无法创建Topic。

2. AdminClient创建失败：

   • AdminClient可能在Kafka容器完全启动就绪前就执行了创建操作，此时Kafka服务尚未就绪，导致连接超时或失败。
   • kafka.producer.properties中的配置（如安全协议）与Kafka容器默认配置不匹配，导致无法建立连接。

解决方法 #

方法1：使用Testcontainers KafkaContainer内置的Topic创建方法 #

Testcontainers提供了withTopics方法，无需手动设置环境变量，兼容性更好：

@Bean(initMethod = "start", destroyMethod = "stop")
public KafkaContainer kafkaContainer(KafkaContainerConfiguration kafkaContainerConfiguration,
                                     ZookeeperContainer zookeeperContainer) {
    return new KafkaContainer(KAFKA_IMAGE_NAME)
            .withNetwork(network)
            .withTopics("subs-topic", "subs-topic-notifications") // 直接指定要创建的Topic，默认1分区1副本
            .withNetworkAliases(kafkaContainerConfiguration.getBootstrapHost())
            .withStartupTimeout(Duration.ofSeconds(180))
            .withExternalZookeeper(zookeeperContainer.getNetworkAliases().get(0) + ":2181"); // 明确指定Zookeeper默认端口2181
}

方法2：确保AdminClient在Kafka就绪后执行创建操作 #

如果必须使用AdminClient，添加一个初始化Bean等待Kafka就绪后再创建Topic：

@Configuration
public class TopicInitializer {

    @Bean
    public CommandLineRunner createTopics(AdminClient adminClient,
                                          KafkaContainer kafkaContainer,
                                          KafkaContainerConfiguration config) {
        return args -> {
            // 等待Kafka容器完全启动就绪
            kafkaContainer.waitingFor(Wait.forLogMessage(".*Kafka Server started.*", 1));
            
            // 定义要创建的Topic
            NewTopic dataTopic = new NewTopic(config.getDataTopic(), 1, (short) 1);
            NewTopic notificationTopic = new NewTopic(config.getNotificationTopic(), 1, (short) 1);
            
            // 执行创建并等待完成
            adminClient.createTopics(List.of(dataTopic, notificationTopic)).all().get();
        };
    }
}

关键检查点 #

• 确认CLIENT_PORT常量值为Zookeeper默认客户端端口2181，避免Kafka无法连接Zookeeper。
• 检查kafka.producer.properties，确保未配置SSL等安全认证（默认Testcontainers Kafka容器无认证）。
• 保留足够的启动超时时间（已设置180秒，满足需求），避免Kafka启动慢导致后续操作失败。

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