如何解决卡夫卡消费者错误处理偏移量重置
我正在使用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
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。