如何在反应器kafka中重试失败的消费者记录

Question

我正在尝试使用reamer-kafka来消耗消息。其他一切都很好，但我想为失败的消息添加重试（2）。spring-kafka已经默认重试失败记录3次，我想使用reamer-kafka实现相同。

我用SpringKafka作为反应Kafka的包装。以下是我的消费者模板：

reactiveKafkaConsumerTemplate
                .receiveAutoAck()
                .map(ConsumerRecord::value)
                .flatMap(this::consumeWithRetry)
                .onErrorContinue((error, value)->log.error("something bad happened while consuming : {}", error.getMessage()))
                .retryWhen(Retry.backoff(30, Duration.of(10, ChronoUnit.SECONDS)))
                .subscribe();

让我们考虑消耗方法如下

public Mono<Void> consume(MessageRecord message){
   return Mono.error(new RuntimeException("test retry"); //sample error scenario
}

我使用以下逻辑在失败时重试消耗方法。

public Mono<Void> consumeWithRetry(MessageRecord message){
   return consume(message)
          .retry(2);
}

如果当前消费者记录异常失败，我想重试使用该消息。我试图用另一次重试（3）来包装consume方法，但这并没有达到目的。最后一次retryWhen仅用于重试Kafka再平衡订阅。

@simon-baslé@加里-罗素

Answer 1

之前在重试时，我使用了以下方法：

public Mono<Void> consumeWithRetry(MessageRecord message){
   return consume(message)
          .retry(2);
}

但它没有重试。添加Mono.defer后，上面的代码工作并添加所需的重试。

public Mono<Void> consumeWithRetry(MessageRecord message){
   return Mono.defer(()->consume(message))
          .retry(2);
}