如何确认Kafka主题已被彻底删除？

Kafka的Admin.deleteTopics API仅在删除请求被集群接收后就返回，此时主题只是被标记为待删除状态，实际删除操作是异步执行的。这就导致如果立刻尝试重建同名主题，会抛出TopicExistsException，而Admin.listTopics()在提交删除请求后就无法再查到该主题，无法判断删除是否真正完成。

可行的编程方案（优先使用官方API） #

1. 轮询调用Admin.describeTopics()检查

主题处于待删除状态时，describeTopics()仍然能查询到该主题的信息；只有当主题被完全删除后，调用该API才会抛出UnknownTopicOrPartitionException。基于这个特性，我们可以通过轮询的方式判断删除是否完成：

private boolean isTopicFullyDeleted(Admin admin, String topicName) {
    try {
        // 尝试查询主题信息
        admin.describeTopics(Collections.singleton(topicName)).all().get();
        return false; // 能查到，说明仍处于待删除状态
    } catch (ExecutionException e) {
        if (e.getCause() instanceof UnknownTopicOrPartitionException) {
            return true; // 抛出该异常，说明主题已完全删除
        }
        // 其他异常按业务需求处理
        throw new RuntimeException("检查主题删除状态失败", e);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new RuntimeException("检查主题删除状态被中断", e);
    }
}

// 使用示例
final var newTopic = new NewTopic("aaa", Optional.empty(), Optional.empty());
CreateTopicsOptions opt = new CreateTopicsOptions();

// 创建主题
admin.createTopics(Collections.singleton(newTopic), opt).all().get();
// 提交删除请求
admin.deleteTopics(Arrays.asList("aaa")).all().get();

// 轮询等待主题完全删除
long timeout = System.currentTimeMillis() + 30000; // 设置30秒超时
while (!isTopicFullyDeleted(admin, "aaa")) {
    if (System.currentTimeMillis() > timeout) {
        throw new RuntimeException("主题删除超时");
    }
    Thread.sleep(1000); // 每秒轮询一次，可根据集群性能调整间隔
}

// 此时可以安全重建主题
admin.createTopics(Collections.singleton(newTopic), opt).all().get();

注意事项

• 轮询间隔：根据集群的性能调整，避免过于频繁的API调用给集群带来额外负载
• 超时设置：必须设置合理的超时时间，防止程序无限等待
• 异常处理：除了UnknownTopicOrPartitionException，其他异常需要根据业务场景做针对性处理

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