反应堆通量抛出llegalArgumentException-由于bufferTimeout而怀疑

问题描述

我有一个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$1
$1.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中发生，而仅在应用程序的生产环境中发生。 任何线索都将有所帮助。

谢谢！

解决方法

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