Kafka Log Compacted Topic Duplication Values against the same key not deleted答案

【问题标题】：Kafka Log Compacted Topic Duplication Values against same key not deletedKafka Log Compacted Topic Duplication Values against the same key not deleted
【发布时间】：2020-07-23 05:37:57
【问题描述】：

日志压缩主题不应针对相同的键保留重复项。但在我们的例子中，当发送具有相同键的新值时，不会删除前一个值。可能是什么问题？

val TestCompactState: KTable[String, TestCompact] = builder.table[String, TestCompact](kafkaStreamConfigs.getString("testcompact-source"),
   (TestCompactmaterialized).withKeySerde(stringSerde).withValueSerde(TestCompactSerde))

我得到了什么 实际结果

Offsets      Keys        Messages
5            {"id":5}   {"id":5,"namee":"omer","__deleted":"false"}
6            {"id":5}   {"id":5,"namee":"d","__deleted":"false"}

我只想要针对同一个键的最新一条记录 预期结果

6            {"id":5}   {"id":5,"namee":"d","__deleted":"false"}

【问题讨论】：

标签： apache-kafka apache-kafka-streams apache-kafka-connect kafka-producer-api kafka-topic

【解决方案1】：

据我所知，不可能应用日志压缩策略来准确每个键保留一条消息。即使您设置cleanup.policy=compact（主题级别）或log.cleanup.policy=compact（全局级别），也不能保证只会保留最新消息而压缩旧消息。

根据official Kafka documentation：

日志压缩为我们提供了更细粒度的保留机制，以便我们保证至少保留每个主键的最后一次更新

【讨论】：

【解决方案2】：

这种行为可能有多种原因。压缩清理策略不会在每条传入消息之后运行。取而代之的是broker configuration

log.cleaner.min.compaction.lag.ms：消息在日志中保持未压缩的最短时间。仅适用于正在压缩的日志。

类型：长；默认值：0；有效值：;更新模式：集群范围

这默认为0，所以这可能不是原因，但值得检查。

请务必注意，compact 策略从不压缩当前段。消息仅适用于非活动段上的压缩。确保验证

log.segment.bytes：单个日志文件的最大大小

类型：int；默认：1073741824；有效值：[14,...]；更新模式：集群范围

压缩通常由日志当前（“脏”）段中的数据触发。术语“脏”来自未清洁/未压缩。还有另一种配置可以帮助控制压实。

log.cleaner.min.cleanable.ratio：符合清理条件的日志的脏日志与总日志的最小比率。如果还指定了 log.cleaner.max.compaction.lag.ms 或 log.cleaner.min.compaction.lag.ms 配置，则日志压缩器认为日志符合压缩条件，只要：(i)已达到脏比率阈值，并且日志至少有 log.cleaner.min.compaction.lag.ms 持续时间的脏（未压缩）记录，或者（ii）如果日志最多有脏（未压缩）记录log.cleaner.max.compaction.lag.ms 周期。

类型：双；默认值：0.5；有效值：;更新模式：集群范围

默认情况下，要压缩的消息的删除延迟非常高，如下面的配置描述所示。

log.cleaner.max.compaction.lag.ms：消息在日志中保持不符合压缩条件的最长时间。仅适用于正在压缩的日志。

类型：长；默认：9223372036854775807；有效值：;更新模式：集群范围

总而言之，您观察所描述的内容可能有几个原因。并且非常重要的是要知道压缩主题确实不提供任何保证具有相同键的重复消息。它只能保证“至少”保留相同密钥的最新消息。

有一个很好的blog 更详细地解释了日志压缩。

【讨论】：

【解决方案3】：

分区的活动段永远不会被压缩，因此在开始删除旧的重复项之前，可能需要一些时间和更多的消息发送到主题。

【讨论】：