apache-kafka - 如何在不删除主题的情况下删除/清理 Kafka 排队消息

Question

有没有办法在不删除Kafka主题的情况下删除队列消息？
我想在激活消费者时删除队列消息。

我知道有几种方法，例如:

重置保留时间
$ ./bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic MyTopic --config retention.ms=1000

删除kafka文件
$ rm -rf /data/kafka-logs/<topic/Partition_name>

Answer 1

在 0.11 或更高版本中，您可以运行 bin/kafka-delete-records.sh命令来标记要删除的消息。

https://github.com/apache/kafka/blob/trunk/bin/kafka-delete-records.sh

比如发布100条消息

seq 100 | ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic mytest

然后用新的 kafka-delete-records.sh 删除这 100 条消息中的 90 条。
命令行工具

./bin/kafka-delete-records.sh --bootstrap-server localhost:9092 --offset-json-file ./offsetfile.json

哪里 offsetfile.json包含

 {"partitions": [{"topic": "mytest", "partition": 0, "offset": 90}], "version":1 }

然后从头开始消费消息以验证 100 条消息中的 90 条确实被标记为已删除。

./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic mytest --from-beginning
91
92
93
94
95
96
97
98
99
100