apache-kafka - 如何为 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