如何在 Kafka 中为压缩主题设置无限保留？

Question

我正在努力让一个紧凑的主题按预期工作。

我有一个压缩主题，并且消息正在正确压缩，但是当旧消息超过默认保留期时，它们会被删除。

我想要一个压缩主题，它至少有一个无限期的键值。

如何做到这一点？我是否必须覆盖该特定压缩主题的保留期？或者压缩主题是否应该默认无限期地保留一个键的一个值？

我的配置

log.cleaner.backoff.ms = 15000
log.cleaner.dedupe.buffer.size = 134217728
log.cleaner.delete.retention.ms = 1000
log.cleaner.enable = true
log.cleaner.io.buffer.load.factor = 0.9
log.cleaner.io.buffer.size = 524288
log.cleaner.io.max.bytes.per.second = 1.7976931348623157E308
log.cleaner.min.cleanable.ratio = 0.001
log.cleaner.min.compaction.lag.ms = 0
log.cleaner.threads = 1
log.cleanup.policy = [compact, delete]

log.retention.bytes = -1
log.retention.check.interval.ms = 5000
log.retention.hours = 0
log.retention.minutes = 20
log.retention.ms = null

Answer 1

我找到了一个我想分享的解决方案。不幸的是，Kafka 文档对此不是很清楚，所以也许这会对某人有所帮助：

不要设置这个：

log.cleanup.policy = [compact, delete]

此设置意味着所有主题都被压缩和删除。因此，您的主题将根据压缩规则进行压缩，但是当段（消息）超过设置的保留时间（在我的情况下为 20 分钟）时，它们也会被删除。

将默认清理策略设置为：

log.cleanup.policy = compact
or 
log.cleanup.policy = delete

（log.cleanup.policy = delete 是默认配置）

"log.cleanup.policy = compact" 表示主题默认设置为压缩。

设置此默认策略后，您无需进行任何更改。无需将 log.retention 设置为 -1 或任何其他值。您的主题将被压缩并且永远不会删除旧消息（根据压缩规则）

"log.cleanup.policy = delete" 表示默认情况下主题将在保留时间之后被修剪。

如果您选择此默认策略，那么您将需要覆盖每个主题的 cleanup.policy；也就是说，在这个主题上明确设置 cleanup.policy=compact。这将使这个特定主题使用压缩，而不是删除。您无需调整 log.retention。

PS，直觉上，我认为当您基于每个主题指定“log.cleanup.policy = compact”时，默认的“log.cleanup.policy = [compact, delete]”会被覆盖，但事实并非如此.使用“log.cleanup.policy = [compact, delete]”，您可以有效地覆盖紧凑主题的工作方式；您将紧凑更改为紧凑+删除。

PS2，如果您在测试和压缩主题方面遇到问题，请注意只能压缩非活动文件段；活动段永远不会被压缩。因此，为了进行测试，请将 log.segment.bytes 设置为较小的值，例如 10000

Answer 2

只选择“compact”作为清理策略，并设置无限保留。

log.cleanup.policy = [compact]
log.retention.bytes = -1
log.retention.ms = -1