Kafka Log Compaction 总是显示相同键的最后两条记录

Question

找到了这两个问题：here和here，但我还是不太明白。 我仍然有（意外？）行为。

我尝试使用此配置记录压缩 kafka 主题

kafka-topics.sh --bootstrap-server localhost:9092 --create --partitions 1 --replication-factor 1 --topic test1 --config "cleanup.policy=compact" --config "delete.retention.ms=1000" --config "segment.ms=1000" --config "min.cleanable.dirty.ratio=0.01" --config "min.compaction.lag.ms=500"

然后我发送这些消息，每个至少有 1 秒的间隔

A: 3
A: 4
A: 5
B: 10
B: 20
B: 30
B: 40
A: 6

我期望的是几秒钟后（配置为 1000？），当我运行 kafka-console-consumer.sh --bootstrap-server localhost:9092 --property print.key=true --topic test1 --from-beginning 时，我应该得到

A: 6
B: 40

相反，我得到了：

A: 5
B: 40
A: 6

如果我发布另一条消息B:50 并运行消费者，我得到：

B: 40
A: 6
B: 50

而不是预期的

A: 6
B: 50

1. 实际上，如何配置日志压缩？
2. 来自Kafka documentation：日志压缩确保 Kafka 始终为单个主题分区的数据日志中的每个消息键至少保留最后一个已知值
   这是否意味着我只能对具有单个分区的主题使用日志压缩？

Answer 1

基本上，您已经自己提供了答案。如 Kafka 文档中所述，“日志压缩确保 Kafka 将始终为单个主题分区的数据日志中的每个消息键保留至少最后一个已知值”。因此，不能保证您将始终为一个键提供一条消息。

如果我正确理解了日志压缩，它并不适用于您在非常有效的问题中提出的用例。相反，它的目的是最终达到主题中每个键只出现一条消息的阶段。

日志压缩是一种机制，可提供更细粒度的每条记录保留，而不是更粗粒度的基于时间的保留。这个想法是有选择地删除我们具有相同主键的最近更新的记录。这样可以保证日志至少具有每个键的最后一个状态。

如果您打算只保留每个键的最新状态，并以尽可能少地处理旧状态为目标，那么压缩主题是正确的选择（非压缩主题的情况取决于时间/大小）基于保留）。据我所知，日志压缩的用例是保留最新的地址、手机号码、数据库中的值等。这些值不会每时每刻都在变化，而且您通常有很多键。

从技术角度来看，我猜你的情况如下。

当涉及到压缩时，日志被视为分为两部分

• 干净：之前已经压缩过的邮件。此部分仅包含每个键的一个值，即前一次压缩时的最新值。
• Dirty：上次压缩后写入的消息。

生成消息B: 40（A: 5 已生成）后，日志的clean 部分为空，dirty/active 部分包含A: 5 和B: 40。消息A: 6 根本不是日志的一部分。生成新消息A: 6 将开始压缩日志的脏部分（因为您的比率非常低），但不包括新消息本身。如前所述，没有什么要清理的了，所以新消息只会被添加到主题中，现在位于日志的脏部分中。您在生成 B: 50 时观察到的情况也是如此。

此外，压缩将永远不会在您的活动段上发生。因此，即使您将segment.ms 设置为仅1000 ms，它也不会产生新的段，因为在产生A: 6 或B: 50 之后没有新数据传入。

要解决您的问题并遵守预期，您需要在生成 A: 6 或 B: 50 后生成另一条消息 C: 1。这样，cleaner 可以再次比较日志的干净和脏部分，并删除A: 5 或B: 40。

与此同时，查看这些段在 Kafka 日志目录中的行为。

在我看来，日志压缩的配置完全没问题！这不是观察预期行为的正确用例。但是对于生产用例，请注意您当前的配置会尝试非常频繁地启动压缩。根据您的数据量，这可能会变得相当 I/O 密集。默认比率设置为 0.50 并且 log.roll.hours 通常设置为 24 小时是有原因的。此外，您通常希望确保消费者在压缩之前有机会读取所有数据。