Spring Kafka倾听所有主题并调整分区偏移

Question

基于spring-kafka的文档,我使用基于注释的@KafkaListener来配置我的消费者.

我看到的是 -

1. 除非我将偏移量指定为零,否则在开始时,Kafka消费者将收集未来的消息,而不是现有的消息.(我知道这是预期的结果,因为我没有指定我想要的偏移量)

2. 我在文档中看到一个选项,用于指定主题+分区组合以及零偏移,但如果我这样做 - 我必须明确指定我希望我的消费者听哪个主题.

使用上面的方法2,这就是我的消费者现在的样子 -

@KafkaListener(id = "{group.id}",
        topicPartitions = {
                @TopicPartition(topic = "${kafka.topic.name}",
                        partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "0"))
        },
        containerFactory = "kafkaListenerContainerFactory")
public void listen(@Payload String payload,
                   Acknowledgment ack) throws InterruptedException, IOException {

    logger.debug("This is what we received in the Kafka Consumer = " + payload);

    idService.process(payload);

    ack.acknowledge();
}

虽然我知道有一个选项可以指定"topicPattern"通配符或"主题"列表作为注释配置的一部分,但我没有看到我可以提供偏移值从零开始的地方列出的主题/主题模式.有没有办法将两者结合起来？请指教.

Answer 1

当使用主题和topicPatterns(而不是明确声明分区)时,Kafka决定哪个消费者实例将获得哪些分区.

Kafka将分配分区,初始偏移量将是该组ID的最后一次提交.您目前无法更改该偏移量,但我们正在考虑添加搜索功能.

如果您始终想要从第一个可用偏移开始,请使用唯一的组ID(例如UUID.randomUUID().toString())并设置

props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

由于Kafka将没有该组ID的现有偏移量,因此它将使用该属性来确定从哪里开始.

您也可以使用MANUAL ack模式,而不是ack,这将有效地做同样的事情.