卡夫卡消费者错误处理偏移量重置

问题描述

我正在使用Spring Boot应用程序2.4版中来自kafka流的事件。 kafka客户端的版本是2.3。有两个使用者在使用事件。如果发生任何错误，我想将事件放回kafka中。我不想将失败的事件放入死信队列。我正在使用ConsumerAwareListenerErrorHandler。

@Override
public Object handleError(Message<?> message,ListenerExecutionFailedException exception,Consumer<?,?> consumer) {

    ConsumerRecord record = (ConsumerRecord) message.getPayload();
    // consumer.seek(new TopicPartition(record.topic(),record.partition()),record.offset());
    Collection collection = Arrays.asList(new TopicPartition(record.topic(),record.partition()));
    consumer.seekToBeginning(collection);

    return null;
}

现在我要停止的使用者是，另一个正在运行的使用者应使用相同的错误事件。请帮助。

谢谢

解决方法

那将行不通，因为前一个poll()提取的任何其他记录仍将被处理；改用SeekToCurrentErrorHandler。

https://docs.spring.io/spring-kafka/docs/2.5.5.RELEASE/reference/html/#seek-to-current