java - 当偏移量存在时，如何消耗kafka frome的最大偏移量？

Question

我已阅读文档并找到了配置“auto.offset.reset”:

当 ZooKeeper 中没有初始偏移量或偏移量超出范围时该怎么办:

问题是我曾经使用组id来消费kafka，我想保留组id但放弃旧消息。

我该怎么做？

Answer 1

您可以尝试在消费者中的 ConsumerRebalanceListener.onPartitionsAssigned 中执行此操作:

public void onPartitionsAssigned(Collection<TopicPartition> partitions) {   
    kafkaConsumer.seekToEnd(partitions);
}