删除 Apache kafka 中的主题消息

Question

我正在测试 kafka-topics 的工作，但我不明白删除是如何工作的。

我创建了一个简单的主题

retention.ms = 60000

和

segment.ms = 60000

和

cleanup.policy=delete.

在此之后，我创建了一个生产者并发送了一些消息。 消费者接收消息没有问题。 但我希望，一分钟后，如果重复消费者，它不会显示消息，因为它们必须已被删除。但是这种行为不会发生。

如果我在 ksql 中创建一个查询，它是一样的。消息总是出现。

我想我不明白删除是如何工作的。

例子：

1) 主题

./kafka-topics --create --zookeeper localhost:2181 --topic test -- 
  replication-factor 2 --partitions 1 --config "cleanup.policy=delete" -- 
  config "delete.retention.ms=60000" --config "segment.ms=60000"

2) 制作人

./kafka-avro-console-producer --broker-list broker:29092 --topic test-- 
  property parse.key=true --property key.schema='{"type":"long"}' --property 
  "key.separator=:" --property value.schema='{"type": "record","name": 
  "ppp","namespace": "test.topic","fields": [{"name": "id","type": "long"}]}'

3) 来自生产者的消息

1:{"id": 1}
 2:{"id": 2}
 4:{"id": 4}
 5:{"id": 5}

4) 消费者

  ./kafka-avro-console-consumer \
    --bootstrap-server broker:29092 \
    --property schema.registry.url=http://localhost:8081 \
    --topic test--from-beginning --property print.key=true

消费者显示四个消息。

但我希望如果我在一分钟后再次运行消费者（我也等待了更多时间，甚至几个小时），消息不会显示，因为retention.ms 和segment.ms 是一分钟。

消息何时被实际删除？

Answer 1

使用kafka-topics --zookeeper localhost:2181 --alter --topic test --config retention.ms=60000 更改上面Ajay Srivastava 提到的retention.ms 并再次测试。

Answer 2

另一个在 Kafka 删除过程中要知道的重要思想是log segment file：

主题分为分区对吗？这就是允许并行性、规模化等的原因。

每个分区分为log segments files。为什么？因为 Kafka 将数据写入磁盘，对吧……？我们不想将整个 topic / partition 保存在 1 个大文件中，而是将其拆分为更小的文件（段）..

将数据分成更小的文件有很多优点，与问题无关。可以阅读更多here

这里要注意的关键是：

保留策略正在查看日志 semgnet 的文件时间戳。

"按时间保留是通过检查最后修改的 磁盘上每个日志段文件的时间 (mtime)。在正常情况下 ter 操作，这是日志段关闭的时间，以及 表示文件中最后一条消息的时间戳"

（来自Kafka-definitive Guide，第 26 页）

版本0.10.1.0

日志保留时间不再基于日志段的最后修改时间。相反，它将基于日志段中消息的最大时间戳。

这意味着它只在 关闭 日志段文件上查看。 确保您的“细分”config 参数是正确的..