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

8hhllhi2  于 2021-06-07  发布在  Kafka
关注(0)|答案(2)|浏览(474)

有没有办法删除队列消息而不删除Kafka主题?
我想在激活消费者时删除队列消息。
我知道有以下几种方法:
重置保留时间 $ ./bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic MyTopic --config retention.ms=1000 删除Kafka文件 $ rm -rf /data/kafka-logs/<topic/Partition_name>

c6ubokkw

c6ubokkw1#

在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 命令行工具

./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
3xiyfsfu

3xiyfsfu2#

要删除特定主题中的所有邮件,可以运行 kafka-delete-records.sh 例如,我有一个主题叫做 test ,其中有4个 partitions .
创建 Json 例如,文件 j.json :

{

"partitions": [

    {

        "topic": "test",

        "partition": 0,

        "offset": -1

    }, {

        "topic": "test",

        "partition": 1,

        "offset": -1

    }, {

        "topic": "test",

        "partition": 2,

        "offset": -1

    }, {

        "topic": "test",

        "partition": 3,

        "offset": -1

    }

],

"version": 1

}

现在通过以下命令删除所有消息:

/opt/kafka/confluent-4.1.1/bin/kafdelete-records --bootstrap-server 192.168.XX.XX:9092 --offset-json-file j.json

执行命令后,将显示此消息

Records delete operation completed:
partition: test-0   low_watermark: 7
partition: test-1   low_watermark: 7
partition: test-2   low_watermark: 7
partition: test-3   low_watermark: 7

相关问题