gpt4 book ai didi

java - 所有分区的 seekToEnd 并在 Kafka 消费者的自动重新平衡中幸存下来

转载 作者:行者123 更新时间:2023-11-30 06:47:58 25 4
gpt4 key购买 nike

当消费者组 A 的 Kafka 消费者连接到 Kafka 代理时,我想查找所有分区的末尾,即使在代理端存储了偏移量。如果更多的消费者正在为同一个消费者组连接,他们应该获取最新存储的偏移量。我正在做以下事情:

consumer.poll(timeout) 
consumer.seekToEnd(emptyList())

while(true) {
val records = consumer.poll(timeout)
if(records.isNotEmpty()) {
//print records
consumer.commitSync()
}
}

问题是当我连接消费者组 A 的第一个消费者 c1 时,一切都按预期工作,如果我连接消费者组 A 的另一个消费者 c2,该组正在重新平衡,c1 将消耗跳过的偏移量。

有什么想法吗?

最佳答案

您可以创建一个实现ConsumerRebalanceListener 的类,如下所示:

public class AlwaysSeekToEndListener<K, V> implements ConsumerRebalanceListener {

private Consumer<K, V> consumer;

public AlwaysSeekToEndListener(Consumer consumer) {
this.consumer = consumer;
}

@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {

}

@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
consumer.seekToEnd(partitions);
}
}

然后在订阅主题时使用这个监听器:

consumer.subscribe(Collections.singletonList("test"), new AlwaysSeekToEndListener<String, String>(consumer));

关于java - 所有分区的 seekToEnd 并在 Kafka 消费者的自动重新平衡中幸存下来,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/45075147/

25 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com