Kafka独占式消费者的重试行为

在Kafka独占式消费者中，如果出现错误导致消费失败，系统会自动进行重试。但是由于消费者是独占式的，所以重试可能会失败，因为其他消费者已经在处理相同的消息。这将导致消息被永久地标记为失败，从而无法重新处理。为了解决这个问题，可以使用阻塞重试。

阻塞重试是一种在请求失败时进行阻塞的重试策略，而不是立即进行重试。这种策略确保每个请求都有充分的时间处理，这是非常重要的，因为这种阻塞重试策略允许消费者有足够的时间去处理消息。以下是一个使用阻塞重试的示例代码：

ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord record : records) {
    try {
        process(record); // 处理消息
        consumer.commitSync(); // 成功处理，提交并确认
    } catch (Exception e) {
        consumer.pause(Collections.singleton(topicPartition)); // 暂停分区
        while (true) {
            consumer.seek(topicPartition, record.offset()); // 重置消费者位置
            records = consumer.poll(Duration.ofMillis(100)); // 再次读取消息
            try {
                process(record); // 再次处理消息
                consumer.commitSync(); // 成功处理，提交并确认
                consumer.resume(Collections.singleton(topicPartition)); // 恢复分区
                break; // 跳出循环
            } catch (Exception re) {
                // 消息处理失败，进入下一轮重试
            }
        }
    }
}

在这个示例代码中，如果处理消息出现了错误，程序会暂停当前分区，然后使用阻塞重试对消息进行再次处理。在每次重试之间，程序都将重新定位到之前的消息位置，保证每个消息都得到了充分的处理。如果重试成功，程序会恢复消费分区并提交消息。如果重试失败，程序会进入下一轮重试，直到成功为止。