如何使用confluent-kafka-python删除主题

Question

我正在使用Kafka来实现多个微服务之间的通信。服务是用Python编写的，我使用Confluent库来处理Kafka。在某种程度上，我知道一些主题只是“结束”，这样我就可以自动清理它们。

多亏了Confluent库，有没有办法删除“主题”？我找不到任何关于这方面的文件...

谢谢

Answer 1

您可以使用confluent Admin Api's to delete a topic

示例

获取一个AdminClient实例和一个主题列表

def example_delete_topics(a, topics):
    """ delete topics """

    # Call delete_topics to asynchronously delete topics, a future is returned.
    # By default this operation on the broker returns immediately while
    # topics are deleted in the background. But here we give it some time (30s)
    # to propagate in the cluster before returning.
    #
    # Returns a dict of <topic,future>.
    fs = a.delete_topics(topics, operation_timeout=30)

    # Wait for operation to finish.
    for topic, f in fs.items():
        try:
            f.result()  # The result itself is None
            print("Topic {} deleted".format(topic))
        except Exception as e:
            print("Failed to delete topic {}: {}".format(topic, e))

Answer 2

我使用这个简单的Python函数：

def delete_kafka_topic(topic_name):
    call(["/usr/bin/kafka-topics", "--zookeeper", "zookeeper-1:2181", "--delete", "--topic", topic_name])

我的团队在自动化测试中使用它，我们希望能够重新运行测试，验证结果，而不是从以前的测试尝试中看到结果。