apache-kafka - Kafka 日志压缩始终显示同一键的最后两条记录

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 始终为单个主题分区的数据日志中的每个消息键保留至少最后一个已知值”。因此，不能保证您始终能够获得一键对应的一条消息。

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

Log compaction is a mechanism to give finer-grained per-record retention, rather than the coarser-grained time-based retention. The idea is to selectively remove records where we have a more recent update with the same primary key. This way the log is guaranteed to have at least the last state for each key.

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

从技术角度来看，我猜您的情况发生了以下情况。

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

• 干净:之前已压缩的消息。此部分仅包含每个键的一个值，该值是上一次压缩时的最新值。
• 脏消息:上次压缩后写入的消息。

生成消息后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 .

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

同时，看看这些段在 Kafka 日志目录中的行为如何。

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