java - 卡夫卡只订阅最新消息？

Question

有时(似乎很随机)Kafka 会发送旧消息。我只想要最新的消息，所以它会用相同的键覆盖消息。目前看起来我有多条消息具有相同的 key ，它没有被压缩。

我在主题中使用此设置:

cleanup.policy=compact

我正在使用 Java/Kotlin 和 Apache Kafka 1.1.1 客户端。

Properties(8).apply {
    val jaasTemplate = "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"%s\" password=\"%s\";"
    val jaasCfg = String.format(jaasTemplate, Configuration.kafkaUsername, Configuration.kafkaPassword)
    put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
            BOOTSTRAP_SERVERS)
    put(ConsumerConfig.GROUP_ID_CONFIG,
            "ApiKafkaKotlinConsumer${Configuration.kafkaGroupId}")
    put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
            StringDeserializer::class.java.name)
    put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
            StringDeserializer::class.java.name)

    put("security.protocol", "SASL_SSL")
    put("sasl.mechanism", "SCRAM-SHA-256")
    put("sasl.jaas.config", jaasCfg)
    put("max.poll.records", 100)
    put("receive.buffer.bytes", 1000000)
}

我错过了一些设置吗？

Answer 1

如果您希望每个键只有一个值，则必须使用 KTable<K,V>摘要:StreamsBuilder::table(final String topic)来自 卡夫卡流 .此处使用的主题应将清理策略设置为 compact .

如果您使用 KafkaConsumer，您只需从经纪人那里提取数据。它没有为您提供任何执行某种重复数据删除的机制。根据是否执行了压缩，您可以获得相同 key 的一到 n 条消息。

关于压实

压缩并不意味着立即删除同一键的所有旧值。当old相同键的消息将被删除，这取决于几个属性。最重要的是:

log.cleaner.min.cleanable.ratio

The minimum ratio of dirty log to total log for a log to eligible for cleaning

log.cleaner.min.compaction.lag.ms

The minimum time a message will remain uncompacted in the log. Only applicable for logs that are being compacted.

log.cleaner.enable

Enable the log cleaner process to run on the server. Should be enabled if using any topics with a cleanup.policy=compact including the internal offsets topic. If disabled those topics will not be compacted and continually grow in size.

更多关于压实的细节你可以找到 https://kafka.apache.org/documentation/#compaction