Spring、Kafka与Testcontainers集成测试：主题管理及随机错误规避

1. 避免创建/删除主题时的异常 #

Kafka Admin API自带选项可直接忽略不存在/已存在的主题，无需提前手动判断：

• 删除操作：用DeleteTopicsOptions#onlyIfExists(true)，主题不存在时不会抛出异常
• 创建操作：用CreateTopicsOptions#onlyIfNotExists(true)，主题已存在时不会抛出异常

2. 解决删除未完成就创建的并发问题 #

Kafka的主题操作是异步执行的，必须等待操作的Future完成后再执行下一步：

• 调用deleteTopics()后，通过Future#get()阻塞等待删除操作彻底完成
• 创建主题时同样等待createTopics()的Future完成，确保主题已就绪

3. 规避主题非预期为空的随机错误 #

• 删除主题后，可循环检查主题列表，确认主题已被移除再执行创建
• 创建完成后，检查主题的分区leader是否就绪，避免测试提前发送消息导致异常

优化后的主题重建代码 #

public void recreateTopic(String topic) throws ExecutionException, InterruptedException {
    try (Admin admin = Admin.create(properties)) {
        // 删除主题（忽略不存在的情况，设置超时）
        DeleteTopicsOptions deleteOpts = new DeleteTopicsOptions()
                .onlyIfExists(true)
                .timeout(Duration.ofSeconds(10));
        admin.deleteTopics(List.of(topic), deleteOpts).all().get();

        // 确认主题已被删除（循环重试，最多5次）
        int retryCount = 0;
        while (retryCount < 5) {
            Collection<TopicListing> topics = admin.listTopics().listings().get();
            boolean exists = topics.stream().map(TopicListing::name).anyMatch(topic::equals);
            if (!exists) break;
            Thread.sleep(500);
            retryCount++;
        }

        // 创建主题（忽略已存在的情况，设置超时）
        NewTopic newTopic = new NewTopic(topic, 1, (short) 1); // 根据测试需求调整分区、副本数
        CreateTopicsOptions createOpts = new CreateTopicsOptions()
                .onlyIfNotExists(true)
                .timeout(Duration.ofSeconds(10));
        admin.createTopics(List.of(newTopic), createOpts).all().get();

        // 确认主题分区就绪（避免未初始化完成就使用）
        retryCount = 0;
        while (retryCount < 5) {
            DescribeTopicsResult result = admin.describeTopics(List.of(topic));
            Map<String, TopicDescription> descMap = result.all().get();
            TopicDescription desc = descMap.get(topic);
            if (desc != null && desc.partitions().stream().allMatch(p -> p.leader() != null)) {
                break;
            }
            Thread.sleep(500);
            retryCount++;
        }
    }
}

额外测试隔离建议 #

• 给每个测试类/方法的主题添加唯一标识（比如随机后缀、测试类名前缀），从根源避免主题冲突
• 使用Testcontainers的@Container注解，让每个测试类启动独立的Kafka容器，彻底隔离测试环境

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