kafka メッセージの削除

kafka を実験しているとトピックを削除するのでなくメッセージをお掃除したいときがある。kafka-delete-records.shでできるようだ。

削除前の確認

パーティション毎のメッセージ確認

# bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.21:9092 --topic partitioned-topic --from-beginning --partition 0
b
e
^CProcessed a total of 2 messages
# bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.21:9092 --topic partitioned-topic --from-beginning --partition 1
c
f
^CProcessed a total of 2 messages
# bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.21:9092 --topic partitioned-topic --from-beginning --partition 2
a
d
^CProcessed a total of 2 messages

削除

offsetfile.json を作成。削除というか -1 でoffsetを最後まで進めている。

{"partitions":[{"topic":"partitioned-topic","partition":0,"offset": -1}],"version":1}

# bin/kafka-delete-records.sh --bootstrap-server 192.168.0.21:9092 --offset-json-file offsetfile.json
Executing records delete operation
Records delete operation completed:
partition: partitioned-topic-0  low_watermark: 15

削除後の確認

# bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.21:9092 --topic partitioned-topic --from-beginning --partition 0
^CProcessed a total of 0 messages
# bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.21:9092 --topic partitioned-topic --from-beginning --partition 1
c
f
^CProcessed a total of 2 messages
# bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.21:9092 --topic partitioned-topic --from-beginning --partition 2
a
d
^CProcessed a total of 2 messages
# bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.21:9092 --topic partitioned-topic --from-beginning
a
d
c
f
^CProcessed a total of 4 messages

参考

stackoverflow.com