Reactor flux 抛出 illegalArgumentException - 怀疑是由于 bufferTimeout

Question

我有一个 spring 应用程序，它构建了一个响应式管道，如下所示：

buildPipeline(). // returns a flux based on changeStreamEvents or Kafka receives
.bufferTimeout( capacity, Duration.ofSeconds(1))
. flatMap( r -> {
   element x = r.get(r.size()-1)
   //some processing on element and the batch obtained
})
.doOnError( e-> log.info("error occurred:" + e.toString())
.subscribe()

但是，我看到我的应用程序中间抛出以下错误 -

java.lang.illegalArgumentException:3.9 While the Subscription is not cancelled, Subscription.request(long n) MUST throw a java.lang.illegalArgumentException if argument <= 0
at com.mongodb.reactivestreams.client.internal.ObservableToPublisher
.request(ObservableToPublisher.java:43)
at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:155)
at reactor.core.publisher.FluxBufferTimeout
$BufferimeoutSubscriber.requestMore(FluxBufferTimeout.java:317)

我无法确定哪里出了问题，以及为什么流会因该错误而终止。任何帮助将不胜感激。

在我添加“bufferTimeout”以添加批处理功能后，应用程序开始抛出此错误。在此之前，我从未遇到过这种异常。也不确定如何复制该问题，因为它不是在本地或 UAT 中发生，而是仅在应用程序的生产环境中发生。任何线索都会有所帮助。

谢谢！

Answer 1

尝试添加 onBackPressureBuffer()，以便在需求较低的情况下此运算符缓冲请求，并以受控方式发出项目。

Reactor flux 抛出 illegalArgumentException - 怀疑是由于 bufferTimeout

Reactor flux throws illegalArgumentException - suspecting due to bufferTimeout

project-reactor

spring-mongodb

changestream

spring-data-mongodb-reactive