Как удалить/очистить сообщения Kafka в очереди без удаления темы

Есть ли способ удалить сообщения в очередь без удаления тем Кафки?
Я хочу удалить сообщения очереди при активации пользователя.

Я знаю, что существует несколько способов:

  1. Сброс времени удерживания

    $./bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic MyTopic --config retention.ms=1000

  2. Удаление файлов kafka

    $ rm -rf/data/kafka-logs/<topic/Partition_name>

Ответы

Ответ 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

затем удалите 90 из этих 100 сообщений с помощью нового инструмента командной строки 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 }

и затем использовать сообщения с самого начала, чтобы убедиться, что 90 из 100 сообщений действительно помечены как удаленные.

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

Ответ 2

Чтобы удалить все сообщения в определенной теме, вы можете запустить 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