通过 Java KafkaServer API 删除和重新创建 Kafka 主题答案

【问题标题】：Deleting and re-creating Kafka topics through the Java KafkaServer API通过 Java KafkaServer API 删除和重新创建 Kafka 主题
【发布时间】：2016-05-03 12:55:52
【问题描述】：

我对连接到本地 Kafka 实例的应用程序进行了一些集成测试。当测试以类似于该问题接受的答案的方式运行时，我正在使用 Java KafkaServer API 按需创建本地实例：

How can I instanciate a Mock Kafka Topic for junit tests?

我的每个测试都在单独运行时通过。我遇到的问题是我的测试使用相同的 Kafka 主题，我希望主题开始每个不包含消息的测试。但是，当我连续运行测试时，当第一次运行后的所有测试并尝试重新创建他们需要的主题时，我会收到此错误：

kafka.common.TopicExistsException: Topic "test_topic" already exists.
    at kafka.admin.AdminUtils$.createOrUpdateTopicPartitionAssignmentPathInZK(AdminUtils.scala:187)
    at kafka.admin.AdminUtils$.createTopic(AdminUtils.scala:172)
    at kafka.admin.TopicCommand$.createTopic(TopicCommand.scala:93)

每个测试都会创建和关闭自己的 EmbeddedZookeeper 和 KafkaServer。我还尝试在每次测试结束时从 ZK 以及 KafkaServer 的 logDirs 中删除“经纪人/主题”路径。不知何故，第一次测试的主题仍然存在到第二次。

在每个测试结束时我可以做些什么来确保它使用的主题不会干扰在它之后运行的测试？

【问题讨论】：

我还尝试在代理上启用“delete.topic.enable”并发送 TopicCommand.deleteTopic 命令。日志报告删除成功，但错误仍在继续。
如果您能提供您正在使用的 kafka 版本，我可能会有所帮助

标签： apache-kafka

【解决方案1】：

我最终能够让它工作。

我没有在每次测试后进行清理，而是将测试更改为在运行前进行清理。

我需要执行两个清理步骤。

首先是在启动KafkaServer之前删除broker的数据目录。

    String dataDirectory = 'tmp/kafka'
    FileUtils.deleteDirectory(FileUtils.getFile(dataDirectory))

    Properties props = TestUtils.createBrokerConfig(BROKER_ID, port, true)
    props.put('log.dir', dataDirectory)
    props.put('delete.topic.enable', 'true')

    KafkaConfig config = new KafkaConfig(props)
    Time mock = new MockTime()
    kafkaServer = TestUtils.createServer(config, mock)

第二种是在Zookeeper中递归删除主题路径，然后再发送createTopic命令。

    zkClient.deleteRecursive(ZkUtils.getTopicPath(topicName))

    List<String> arguments = ['--topic', topicName, '--partitions', '1', '--replication-factor', '1']
    TopicCommand.createTopic(zkClient, new TopicCommand.TopicCommandOptions(arguments as String[]))

我尝试了许多类似的方法，但除了这个之外，我无法让它与任何东西一起使用。

请注意，代码是 Groovy 而不是 Java。

【讨论】：