如何在reactor-kafka中重试失败的ConsumerRecord

Question

我正在尝试reactor-kafka消费消息。其他一切都很好，但我想为失败的消息添加一个retry(2)。默认情况下，spring-kafka已经重试了3次失败的记录，我想用reactor-kafka来实现同样的效果。

我使用spring-kafka作为reactive-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();

让我们考虑一下us方法，如下所示

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

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

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

如果当前的消费者记录出现异常而失败，我想重试消费消息。我试图用另一个retry(3)来包装the方法，但这并不能达到目的。最后一个retryWhen仅用于重试订阅kafka再平衡。

@simon-baslé@gary-russell

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);
}