如何使用Kafka消费者“寻求”正确地返回到所有分区的未提交偏移？

Question

随着JavaKafka消费者寻求（）它需要我们传入TopicPartion和Offest。但是，我认为这个查找方法会为我的消费者收集订阅的Topic分区。

这是我试图处理的例子。

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

Answer 1

我处理一些消息，但我的应用程序出现异常。我不调用commitSync（）

如果不调用commitSync（），则不会提交消息。如果假设异常终止了您的程序，则在重新启动后，使用者通常会从最后提交的偏移量读取该异常。

您可能还需要检查自动。抵消重置并将其设置为最早。

检查您的邮件是否自动提交，因为您正在执行commitSync（）您不需要自动提交，即启用。汽车提交可以设置为false（在合流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（记录，3），其中记录对应于消费者记录，批次重试3次。我认为没有寻求的必要。

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

例如，当我们不使用订阅，即consumer.subscribe（），而是使用consumer.assign（），我们通常只是想偷看（查看）主题中的消息，例如控制台消费者。有时，我们可能需要在一定的偏移量后看到一些消息，或者最后n条消息等，而不需要实际对它们做任何事情，只是显示。