WebFlux反应器kafka重试机制未调用流程方法

Question

我可以知道我的代码中有什么问题吗，因为重试机制没有按预期工作。在我的理解中，每次重试都应该调用process Record方法。

@EventListener(ApplicationReadyEvent.class)
    public void processEvent() {
        log.info("processEvent:: started..");

        Scheduler scheduler = Schedulers.newBoundedElastic(threadPoolConfig.getThreadCap(),
                threadPoolConfig.getQueuedTaskCap(), threadPoolConfig.getThreadPrefix(), threadPoolConfig.getTtlSeconds());

        Flux<ReceiverRecord<String, String>> receiverRecordFlux = Flux.defer(requestReactiveKafkaConsumerTemplate::receive);
        receiverRecordFlux.groupBy(m -> m.receiverOffset().topicPartition())
                .doOnNext(partitionFlux -> log.info("processEvent:: topicPartition {}", partitionFlux.key()))
                .flatMap(partitionFlux -> partitionFlux.subscribeOn(scheduler)
                        .doOnNext(r -> log.info("processEvent:: Record received from offset {} from topicPartition {} with message key {}", r.receiverOffset().topicPartition(), r.key(), r.offset()))
                        .flatMap(this::processRecord)
                        .doOnNext(receiverRecordInfo -> log.info("processEvent:: Record processed from offset {} from topicPartition {} with message key {}", receiverRecordInfo.receiverOffset().offset(), receiverRecordInfo.receiverOffset().topicPartition()))
                        .retryWhen(Retry.backoff(3, Duration.ofMillis(200))
                                .jitter(0d)
                                .doAfterRetry(retrySignal -> log.error("Retried {}", retrySignal.totalRetries()))
                                .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> new KafkaRetryExhaustException())
                        )
                        .doOnError(KafkaRetryExhaustException.class, (msg) -> {
                            log.error("exception... ", msg);
                        })
                        .onErrorResume(throwable -> Mono.empty())
                )
                .subscribe(
                        key -> log.info("Successfully consumed messages, key {}", key),
                        error -> log.error("Error while consuming messages ", error));
    }

原木

 2022-06-06 13:50:33,867 INFO  [reactive-kafka-consumergroupId-1] reactor.util.Loggers$Slf4JLogger: processEvent:: topicPartition test-topic-0
2022-06-06 13:50:33,879 INFO  [CasprConsumer-1] reactor.util.Loggers$Slf4JLogger: processEvent:: Record received from offset test-topic-0 from topicPartition 937 with message key 45
2022-06-06 13:50:33,883 INFO  [CasprConsumer-1] reactor.util.Loggers$Slf4JLogger: processRecord:: processing started
2022-06-06 13:50:33,918 INFO  [CasprConsumer-1] reactor.util.Loggers$Slf4JLogger: processRecord:: is httpRequest null? false
2022-06-06 13:50:34,180 ERROR [reactor-http-nio-3] reactor.util.Loggers$Slf4JLogger: processRecord:: exception.. 
org.springframework.web.reactive.function.client.WebClientResponseException$NotFound: 404 Not Found from POST http://localhost:8080/caps-app/capservices/fulfillment/accesspoints/v3/slots
    at org.springframework.web.reactive.function.client.WebClientResponseException.create(WebClientResponseException.java:202)
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
    *__checkpoint ⇢ 404 from POST http://localhost:8080/caps-app/capservices/fulfillment/accesspoints/v3/slots [DefaultWebClient]
Original Stack Trace:
        at org.springframework.web.reactive.function.client.WebClientResponseException.create(WebClientResponseException.java:202)
        at org.springframework.web.reactive.function.client.DefaultClientResponse.lambda$createException$1(DefaultClientResponse.java:207)
        at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:106)
        at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:79)
        at reactor.core.publisher.FluxDefaultIfEmpty$DefaultIfEmptySubscriber.onNext(FluxDefaultIfEmpty.java:101)
        at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:129)
        at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onNext(FluxContextWrite.java:107)
        at reactor.core.publisher.FluxMapFuseable$MapFuseableConditionalSubscriber.onNext(FluxMapFuseable.java:299)
        at reactor.core.publisher.FluxFilterFuseable$FilterFuseableConditionalSubscriber.onNext(FluxFilterFuseable.java:337)
        at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1816)
        at reactor.core.publisher.MonoCollect$CollectSubscriber.onComplete(MonoCollect.java:159)
        at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:144)
        at reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:260)
        at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:144)
        at reactor.netty.channel.FluxReceive.onInboundComplete(FluxReceive.java:400)
        at reactor.netty.channel.ChannelOperations.onInboundComplete(ChannelOperations.java:419)
        at reactor.netty.channel.ChannelOperations.terminate(ChannelOperations.java:473)
        at reactor.netty.http.client.HttpClientOperations.onInboundNext(HttpClientOperations.java:703)
        at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:93)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
        at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
        at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
        at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:327)
        at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:299)
        at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
        at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
        at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
        at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
        at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:722)
        at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:658)
        at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:584)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:496)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:995)
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.base/java.lang.Thread.run(Thread.java:833)
2022-06-06 13:50:34,391 ERROR [parallel-2] reactor.util.Loggers$Slf4JLogger: Retried 0
2022-06-06 13:50:34,797 ERROR [parallel-3] reactor.util.Loggers$Slf4JLogger: Retried 1
2022-06-06 13:50:35,611 ERROR [parallel-4] reactor.util.Loggers$Slf4JLogger: Retried 2
2022-06-06 13:50:35,613 ERROR [CasprConsumer-1] reactor.util.Loggers$Slf4JLogger: exception... 
com.walmart.caspr.exception.KafkaRetryExhaustException: null
    at com.walmart.caspr.service.ReactiveConsumerService.lambda$processEvent$5(ReactiveConsumerService.java:71)
    at reactor.util.retry.RetryBackoffSpec.lambda$generateCompanion$4(RetryBackoffSpec.java:557)
    at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:376)
    at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onNext(FluxConcatMap.java:251)
    at reactor.core.publisher.EmitterProcessor.drain(EmitterProcessor.java:491)
    at reactor.core.publisher.EmitterProcessor.tryEmitNext(EmitterProcessor.java:299)
    at reactor.core.publisher.SinkManySerialized.tryEmitNext(SinkManySerialized.java:100)
    at reactor.core.publisher.InternalManySink.emitNext(InternalManySink.java:27)
    at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.onError(FluxRetryWhen.java:190)
    at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:222)
    at reactor.core.publisher.FluxFlatMap$FlatMapMain.checkTerminated(FluxFlatMap.java:842)
    at reactor.core.publisher.FluxFlatMap$FlatMapMain.drainLoop(FluxFlatMap.java:608)
    at reactor.core.publisher.FluxFlatMap$FlatMapMain.drain(FluxFlatMap.java:588)
    at reactor.core.publisher.FluxFlatMap$FlatMapMain.onError(FluxFlatMap.java:451)
    at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:222)
    at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.onError(FluxSubscribeOn.java:157)
    at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.subscribe(FluxGroupBy.java:721)
    at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.run(FluxSubscribeOn.java:194)
    at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84)
    at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
    at java.base/java.lang.Thread.run(Thread.java:833)

Answer 1

下面的帖子已经解决了上述问题。重试工作热切而不是懒惰，因此出现了问题。

如何重试reactor-kafka中失败的用户记录