ReactiveElasticsearchClient - Pending acquire queue has reached its maximum size of 1000

[Leuteris]

Executing performance testing on the following method, WebClientRequestException is thrown:

Pending acquire queue has reached its maximum size of 1000; nested exception is reactor.netty.internal.shaded.reactor.pool.PoolAcquirePendingLimitException: Pending acquire queue has reached its maximum size of 1000

@Autowired
private ReactiveElasticsearchClient client;

public Mono<Void> upsertDocument(String index, String docId, Script script, Map<String, Object> docToInsertAsNew) {
		UpdateRequest updateRequest = new UpdateRequest();
		updateRequest.index(index);
		updateRequest.id(docId);
		updateRequest.script(script);
		updateRequest.upsert(docToInsertAsNew);
		updateRequest.retryOnConflict(3);

		return client.update(updateRequest))
		.then();
}

Exception is gone by adding some retries:

return Mono.defer(() -> client.update(updateRequest))
		.retryWhen(Retry.backoff(2, Duration.ofMillis(100))
				.filter(t -> !isClientError(t))
				.maxBackoff(Duration.ofMillis(1000)))
		.then();

Could you suggest a more appropriate solution?
In case of non-idempotent queries(counter increment), for which error responses is it safe to retry?

Full stacktrace

> org.springframework.web.reactive.function.client.WebClientRequestException: Pending acquire queue has reached its maximum size of 1000; nested exception is reactor.netty.internal.shaded.reactor.pool.PoolAcquirePendingLimitException: Pending acquire queue has reached its maximum size of 1000
> 	at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.lambda$wrapException$9(ExchangeFunctions.java:141)
> 	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
> Assembly trace from producer [reactor.core.publisher.MonoErrorSupplied] :
> 	reactor.core.publisher.Mono.error
> 	org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.wrapException(ExchangeFunctions.java:141)
> Error has been observed at the following site(s):
> 	|_         Mono.error ⇢ at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.wrapException(ExchangeFunctions.java:141)
> 	|_                    ⇢ at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.lambda$exchange$4(ExchangeFunctions.java:106)
> 	|_ Mono.onErrorResume ⇢ at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.exchange(ExchangeFunctions.java:106)
> 	|_           Mono.map ⇢ at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.exchange(ExchangeFunctions.java:107)
> 	|_     Mono.doOnError ⇢ at org.springframework.data.elasticsearch.client.reactive.DefaultWebClientProvider.lambda$createWebClientForSocketAddress$3(DefaultWebClientProvider.java:187)
> 	|_                    ⇢ at org.springframework.web.reactive.function.client.ExchangeFilterFunction.lambda$apply$2(ExchangeFilterFunction.java:75)
> 	|_         checkpoint ⇢ Request to POST http://localhost:9200/index/_doc/YRhZDyZl6CR6afpmOYo7bQ/_update?retry_on_conflict=3&timeout=1m [DefaultWebClient]
> 	|_ Mono.switchIfEmpty ⇢ at org.springframework.web.reactive.function.client.DefaultWebClient$DefaultRequestBodyUriSpec.lambda$exchange$7(DefaultWebClient.java:433)
> 	|_         Mono.defer ⇢ at org.springframework.web.reactive.function.client.DefaultWebClient$DefaultRequestBodyUriSpec.exchange(DefaultWebClient.java:430)
> 	|_       Mono.flatMap ⇢ at org.springframework.web.reactive.function.client.DefaultWebClient$DefaultRequestBodyUriSpec.exchangeToMono(DefaultWebClient.java:397)
> 	|_                    ⇢ at org.springframework.data.elasticsearch.client.reactive.DefaultReactiveElasticsearchClient.lambda$sendRequest$20(DefaultReactiveElasticsearchClient.java:562)
> 	|_       Mono.flatMap ⇢ at org.springframework.data.elasticsearch.client.reactive.DefaultReactiveElasticsearchClient.execute(DefaultReactiveElasticsearchClient.java:501)
> 	|_         Mono.error ⇢ at org.springframework.data.elasticsearch.client.reactive.DefaultReactiveElasticsearchClient.lambda$execute$17(DefaultReactiveElasticsearchClient.java:509)
> 	|_ Mono.onErrorResume ⇢ at org.springframework.data.elasticsearch.client.reactive.DefaultReactiveElasticsearchClient.execute(DefaultReactiveElasticsearchClient.java:502)
> 	|_          Flux.from ⇢ at org.springframework.data.elasticsearch.client.reactive.DefaultReactiveElasticsearchClient.sendRequest(DefaultReactiveElasticsearchClient.java:562)
> 	|_                    ⇢ at org.springframework.data.elasticsearch.client.reactive.DefaultReactiveElasticsearchClient.sendRequest(DefaultReactiveElasticsearchClient.java:554)
> 	|_          Flux.next ⇢ at org.springframework.data.elasticsearch.client.reactive.DefaultReactiveElasticsearchClient.update(DefaultReactiveElasticsearchClient.java:368)
> 	|_                    ⇢ at org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient.update(ReactiveElasticsearchClient.java:306)
> Stack trace:
> 		at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.lambda$wrapException$9(ExchangeFunctions.java:141)
> 		at reactor.core.publisher.MonoErrorSupplied.subscribe(MonoErrorSupplied.java:70)
> 		at reactor.core.publisher.Mono.subscribe(Mono.java:4150)
> 		at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onError(FluxOnErrorResume.java:103)
> 		at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onError(FluxPeekFuseable.java:234)
> 		at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onError(FluxPeekFuseable.java:234)
> 		at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onError(FluxPeekFuseable.java:234)
> 		at reactor.core.publisher.MonoNext$NextSubscriber.onError(MonoNext.java:93)
> 		at reactor.core.publisher.MonoFlatMapMany$FlatMapManyMain.onError(MonoFlatMapMany.java:204)
> 		at reactor.core.publisher.SerializedSubscriber.onError(SerializedSubscriber.java:124)
> 		at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.whenError(FluxRetryWhen.java:224)
> 		at reactor.core.publisher.FluxRetryWhen$RetryWhenOtherSubscriber.onError(FluxRetryWhen.java:273)
> 		at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:413)
> 		at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onNext(FluxConcatMap.java:250)
> 		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:97)
> 		at reactor.core.publisher.InternalManySink.emitNext(InternalManySink.java:27)
> 		at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.onError(FluxRetryWhen.java:189)
> 		at reactor.core.publisher.MonoCreate$DefaultMonoSink.error(MonoCreate.java:189)
> 		at reactor.netty.http.client.HttpClientConnect$MonoHttpConnect$ClientTransportSubscriber.onError(HttpClientConnect.java:305)
> 		at reactor.core.publisher.MonoCreate$DefaultMonoSink.error(MonoCreate.java:189)
> 		at reactor.netty.resources.DefaultPooledConnectionProvider$DisposableAcquire.onError(DefaultPooledConnectionProvider.java:173)
> 		at reactor.netty.internal.shaded.reactor.pool.AbstractPool$Borrower.fail(AbstractPool.java:444)
> 		at reactor.netty.internal.shaded.reactor.pool.SimpleDequePool.pendingOffer(SimpleDequePool.java:543)
> 		at reactor.netty.internal.shaded.reactor.pool.SimpleDequePool.doAcquire(SimpleDequePool.java:266)
> 		at reactor.netty.internal.shaded.reactor.pool.AbstractPool$Borrower.request(AbstractPool.java:399)
> 		at reactor.netty.resources.DefaultPooledConnectionProvider$DisposableAcquire.onSubscribe(DefaultPooledConnectionProvider.java:213)
> 		at reactor.netty.internal.shaded.reactor.pool.SimpleDequePool$QueueBorrowerMono.subscribe(SimpleDequePool.java:674)
> 		at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
> 		at reactor.netty.resources.PooledConnectionProvider.lambda$acquire$1(PooledConnectionProvider.java:137)
> 		at reactor.core.publisher.MonoCreate.subscribe(MonoCreate.java:57)
> 		at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
> 		at reactor.netty.http.client.HttpClientConnect$MonoHttpConnect.lambda$subscribe$0(HttpClientConnect.java:269)
> 		at reactor.core.publisher.MonoCreate.subscribe(MonoCreate.java:57)
> 		at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
> 		at reactor.core.publisher.FluxRetryWhen.subscribe(FluxRetryWhen.java:76)
> 		at reactor.core.publisher.MonoRetryWhen.subscribeOrReturn(MonoRetryWhen.java:46)
> 		at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:57)
> 		at reactor.netty.http.client.HttpClientConnect$MonoHttpConnect.subscribe(HttpClientConnect.java:272)
> 		at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
> 		at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
> 		at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
> 		at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:157)
> 		at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120)
> 		at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2397)
> 		at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:162)
> 		at reactor.core.publisher.MonoFlatMap$FlatMapMain.onSubscribe(MonoFlatMap.java:110)
> 		at reactor.core.publisher.FluxMap$MapSubscriber.onSubscribe(FluxMap.java:92)
> 		at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54)
> 		at reactor.core.publisher.Mono.subscribe(Mono.java:4150)
> 		at reactor.core.publisher.FluxFlatMap.trySubscribeScalarMap(FluxFlatMap.java:203)
> 		at reactor.core.publisher.MonoFlatMap.subscribeOrReturn(MonoFlatMap.java:53)
> 		at reactor.core.publisher.Mono.subscribe(Mono.java:4135)
> 		at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.subscribeNext(MonoIgnoreThen.java:255)
> 		at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:51)
> 		at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:157)
> 		at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1815)
> 		at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:249)
> 		at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.complete(MonoIgnoreThen.java:284)
> 		at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.onNext(MonoIgnoreThen.java:187)
> 		at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.subscribeNext(MonoIgnoreThen.java:232)
> 		at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.onComplete(MonoIgnoreThen.java:203)
> 		at reactor.core.publisher.MonoUsingWhen$MonoUsingWhenSubscriber.deferredComplete(MonoUsingWhen.java:280)
> 		at reactor.core.publisher.FluxUsingWhen$CommitInner.onComplete(FluxUsingWhen.java:540)
> 		at reactor.core.publisher.Operators.complete(Operators.java:136)
> 		at reactor.core.publisher.MonoEmpty.subscribe(MonoEmpty.java:45)
> 		at reactor.core.publisher.Mono.subscribe(Mono.java:4150)
> 		at reactor.core.publisher.FluxUsingWhen$UsingWhenSubscriber.onComplete(FluxUsingWhen.java:397)
> 		at reactor.core.publisher.MonoIgnoreElements$IgnoreElementsSubscriber.onComplete(MonoIgnoreElements.java:88)
> 		at reactor.core.publisher.SerializedSubscriber.onComplete(SerializedSubscriber.java:146)
> 		at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.onComplete(FluxRetryWhen.java:199)
> 		at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onComplete(FluxMapFuseable.java:150)
> 		at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onComplete(Operators.java:2057)
> 		at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onComplete(FluxMapFuseable.java:150)
> 		at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1816)
> 		at reactor.core.publisher.MonoCompletionStage.lambda$subscribe$0(MonoCompletionStage.java:82)
> 		at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
> 		at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)
> 		at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
> 		at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2143)
> 		at com.datastax.oss.driver.internal.core.cql.CqlRequestHandler.setFinalResult(CqlRequestHandler.java:321)
> 		at com.datastax.oss.driver.internal.core.cql.CqlRequestHandler.access$1500(CqlRequestHandler.java:94)
> 		at com.datastax.oss.driver.internal.core.cql.CqlRequestHandler$NodeResponseCallback.onResponse(CqlRequestHandler.java:652)
> 		at com.datastax.oss.driver.internal.core.channel.InFlightHandler.channelRead(InFlightHandler.java:257)
> 		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.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286)
> 		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.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
> 		at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:311)
> 		at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:432)
> 		at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
> 		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:719)
> 		at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
> 		at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
> 		at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
> 		at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
> 		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:831)
> Caused by: reactor.netty.internal.shaded.reactor.pool.PoolAcquirePendingLimitException: Pending acquire queue has reached its maximum size of 1000
> 	at reactor.netty.internal.shaded.reactor.pool.SimpleDequePool.pendingOffer(SimpleDequePool.java:543)
> 	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
> Error has been observed at the following site(s):
> 	|_                  ⇢ at reactor.netty.internal.shaded.reactor.pool.SimpleDequePool.acquire(SimpleDequePool.java:110)
> 	|_      Mono.create ⇢ at reactor.netty.resources.PooledConnectionProvider.acquire(PooledConnectionProvider.java:112)
> 	|_                  ⇢ at reactor.netty.tcp.TcpResources.acquire(TcpResources.java:162)
> 	|_                  ⇢ at reactor.netty.http.client.HttpConnectionProvider.acquire(HttpConnectionProvider.java:45)
> 	|_      Mono.create ⇢ at reactor.netty.http.client.HttpClientConnect$MonoHttpConnect.subscribe(HttpClientConnect.java:210)
> 	|_   Flux.concatMap ⇢ at reactor.util.retry.RetrySpec.generateCompanion(RetrySpec.java:356)
> 	|_   Mono.retryWhen ⇢ at reactor.netty.http.client.HttpClientConnect$MonoHttpConnect.subscribe(HttpClientConnect.java:271)
> 	|_                  ⇢ at reactor.netty.http.client.HttpClientFinalizer._connect(HttpClientFinalizer.java:157)
> 	|_ Mono.flatMapMany ⇢ at reactor.netty.http.client.HttpClientFinalizer.responseConnection(HttpClientFinalizer.java:102)
> 	|_        Flux.next ⇢ at org.springframework.http.client.reactive.ReactorClientHttpConnector.connect(ReactorClientHttpConnector.java:113)
> 	|_  Mono.doOnCancel ⇢ at org.springframework.http.client.reactive.ReactorClientHttpConnector.connect(ReactorClientHttpConnector.java:114)
> 	|_ Mono.doOnRequest ⇢ at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.exchange(ExchangeFunctions.java:104)
> 	|_  Mono.doOnCancel ⇢ at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.exchange(ExchangeFunctions.java:105)
> Stack trace:
> 		at reactor.netty.internal.shaded.reactor.pool.SimpleDequePool.pendingOffer(SimpleDequePool.java:543)
> 		at reactor.netty.internal.shaded.reactor.pool.SimpleDequePool.doAcquire(SimpleDequePool.java:266)
> 		at reactor.netty.internal.shaded.reactor.pool.AbstractPool$Borrower.request(AbstractPool.java:399)
> 		at reactor.netty.resources.DefaultPooledConnectionProvider$DisposableAcquire.onSubscribe(DefaultPooledConnectionProvider.java:213)
> 		at reactor.netty.internal.shaded.reactor.pool.SimpleDequePool$QueueBorrowerMono.subscribe(SimpleDequePool.java:674)
> 		at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
> 		at reactor.netty.resources.PooledConnectionProvider.lambda$acquire$1(PooledConnectionProvider.java:137)
> 		at reactor.core.publisher.MonoCreate.subscribe(MonoCreate.java:57)
> 		at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
> 		at reactor.netty.http.client.HttpClientConnect$MonoHttpConnect.lambda$subscribe$0(HttpClientConnect.java:269)
> 		at reactor.core.publisher.MonoCreate.subscribe(MonoCreate.java:57)
> 		at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
> 		at reactor.core.publisher.FluxRetryWhen.subscribe(FluxRetryWhen.java:76)
> 		at reactor.core.publisher.MonoRetryWhen.subscribeOrReturn(MonoRetryWhen.java:46)
> 		at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:57)
> 		at reactor.netty.http.client.HttpClientConnect$MonoHttpConnect.subscribe(HttpClientConnect.java:272)
> 		at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
> 		at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
> 		at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
> 		at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:157)
> 		at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120)
> 		at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2397)
> 		at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:162)
> 		at reactor.core.publisher.MonoFlatMap$FlatMapMain.onSubscribe(MonoFlatMap.java:110)
> 		at reactor.core.publisher.FluxMap$MapSubscriber.onSubscribe(FluxMap.java:92)
> 		at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54)
> 		at reactor.core.publisher.Mono.subscribe(Mono.java:4150)
> 		at reactor.core.publisher.FluxFlatMap.trySubscribeScalarMap(FluxFlatMap.java:203)
> 		at reactor.core.publisher.MonoFlatMap.subscribeOrReturn(MonoFlatMap.java:53)
> 		at reactor.core.publisher.Mono.subscribe(Mono.java:4135)
> 		at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.subscribeNext(MonoIgnoreThen.java:255)
> 		at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:51)
> 		at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:157)
> 		at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1815)
> 		at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:249)
> 		at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.complete(MonoIgnoreThen.java:284)
> 		at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.onNext(MonoIgnoreThen.java:187)
> 		at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.subscribeNext(MonoIgnoreThen.java:232)
> 		at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.onComplete(MonoIgnoreThen.java:203)
> 		at reactor.core.publisher.MonoUsingWhen$MonoUsingWhenSubscriber.deferredComplete(MonoUsingWhen.java:280)
> 		at reactor.core.publisher.FluxUsingWhen$CommitInner.onComplete(FluxUsingWhen.java:540)
> 		at reactor.core.publisher.Operators.complete(Operators.java:136)
> 		at reactor.core.publisher.MonoEmpty.subscribe(MonoEmpty.java:45)
> 		at reactor.core.publisher.Mono.subscribe(Mono.java:4150)
> 		at reactor.core.publisher.FluxUsingWhen$UsingWhenSubscriber.onComplete(FluxUsingWhen.java:397)
> 		at reactor.core.publisher.MonoIgnoreElements$IgnoreElementsSubscriber.onComplete(MonoIgnoreElements.java:88)
> 		at reactor.core.publisher.SerializedSubscriber.onComplete(SerializedSubscriber.java:146)
> 		at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.onComplete(FluxRetryWhen.java:199)
> 		at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onComplete(FluxMapFuseable.java:150)
> 		at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onComplete(Operators.java:2057)
> 		at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onComplete(FluxMapFuseable.java:150)
> 		at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1816)
> 		at reactor.core.publisher.MonoCompletionStage.lambda$subscribe$0(MonoCompletionStage.java:82)
> 		at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
> 		at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)
> 		at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
> 		at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2143)
> 		at com.datastax.oss.driver.internal.core.cql.CqlRequestHandler.setFinalResult(CqlRequestHandler.java:321)
> 		at com.datastax.oss.driver.internal.core.cql.CqlRequestHandler.access$1500(CqlRequestHandler.java:94)
> 		at com.datastax.oss.driver.internal.core.cql.CqlRequestHandler$NodeResponseCallback.onResponse(CqlRequestHandler.java:652)
> 		at com.datastax.oss.driver.internal.core.channel.InFlightHandler.channelRead(InFlightHandler.java:257)
> 		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.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286)
> 		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.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
> 		at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:311)
> 		at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:432)
> 		at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
> 		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:719)
> 		at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
> 		at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
> 		at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
> 		at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
> 		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:831)

[sothawo]

You will have to wait a little for a answer on that, our most capable person with regards to reactive progamming is on vacations; and I'm not this deep into the details.
One question I have: How do you set up the requests for this test?

[Leuteris]

@sothawo I am using JMeter with 500 concurrent users.

[mp911de]

The underlying HTTP connector using Reactor Netty uses pooled connections with a default pool size of 500 and a pending limit size of 1000 (2 * 500). These values are fixed and cannot be configured with the current arrangement.

We do not expose any means to configure ClientHttpConnector or Reactor's HttpClient. Our ReactorClientHttpConnector configuration requires quite some configuration to propagate timeout, proxy and SSL settings. With Spring 5.1, ReactorResourceFactory was introduced to hold and configure HttpClient resources and that seems to be appropriate to configure a ConnectionProvider that has a higher pool size or that doesn't use connection pooling at all.

[spring-projects-issues]

If you would like us to look at this issue, please provide the requested information. If the information is not provided within the next 7 days this issue will be closed.

[Leuteris]

@mp911de what could be the side effects of increasing the pool size?

[mp911de]

A higher number of connections and pending connections with a consequence of higher memory consumption on the client and server-side.