如何正确使用Kafka消费者“seek”来返回所有分区的未提交偏移量？

Question

对于Java Kafka消费者seek()函数，它要求传入topicpartion和offest。但是，我认为这个seek方法将为我的使用者获取订阅的TopicPartitions的集合。

这是我试图处理的例子。

使用者A订阅主题“test-topic”分区1和2。当调用poll()时，我从每个分区读取消息。我处理了一些消息，但我的应用程序得到了一个异常。我不调用commitsync()。现在，我要倒回到在上次poll()中检索到的偏移量，并尝试重新处理它们。那么我该怎么做呢？我是否需要检查每个主题分区最后提交的偏移量，并为每个分区调用seek()？是否多次调用seek()只接受最后一次调用的seek()？正如我所说的，我希望确保我的使用者返回所有分区，这样我就不会丢失任何分配的分区上的任何数据。

Answer 1

我处理了一些消息，但我的应用程序得到了一个异常。我不调用commitSync（）

如果不调用commitsync()，则不会提交消息。如果假设异常杀死了您的程序，那么在重新启动之后，使用者通常从最后提交的偏移量中读取它。

您可能还想检查auto.offset.reset并将其设置为aresty。

检查您的消息是否自动提交，因为您正在执行commitsync()您不需要自动提交，即enable.auto.commit可以设置为false（在Confluent Kafka中默认为true)

如果您的程序没有被异常终止，您总是有消耗的记录。您可以重试处理每个记录，然后提交。

ConsumerRecords records = consumer.poll(Duration.ofSeconds(10));
for(ConsumerRecord record: records)
{
   tryProcess(record, 3);
}
consumer.commitSync();

void tryProcess(ConsumerRecord record, int maxRetries) {
    if(maxRetries < 1) {
        log.warn("max retries exhausted for record");
        return;
    }
    try {
         process(record);
    } catch(Exception ex){ 
        tryProcess(record, --maxRetries);
    }
}

您还可以尝试重试处理记录批，而不是每个记录，如tryprocess(records,3)，其中记录对应于consumerrecords，该批重试3次。我不认为有寻求的必要。

不过，我仍然对seek（）api的用法很好奇

例如，seek()，当我们不使用订阅，即consumer.subscribe()，而使用consumer.assign()，当我们只想查看主题中的消息，例如控制台consumer.subscrime.subscrime.subscrime.subscrime.subscrime.subscrime.subscrime.subscrime有时，我们可能需要在某个偏移量后看到一些消息，或者最后n个消息等，而不需要对它们做任何事情，而只是显示。