pendingacquiretimeout is not working as expected

[manaldush]

It seems that pendingacquiretimeout is not working as expected.
I expected that it is timeout to acquire open connection from pool. So I expected to see timeouts occurred in my unit test in case of connection lack , but timeout is ignored and requests still wait connenction.
In attached log pendingacquiretimeout = 200ms

06/05/2024 19:06:40.584 [DEBUG] reactor.netty.resources.PooledConnectionProvider:pool-4-thread-4 [Loggers.java:259:debug] {} > Creating a new [webclient-pool] client pool [PoolFactory{evictionInterval=PT2S, leasingStrategy=fifo, maxConnections=1, maxIdleTime=300000, maxLifeTime=-1, metricsEnabled=false, pendingAcquireMaxCount=10, **_pendingAcquireTimeout=200_**}] for [localhost:8000]

and some of requests got into pending queue

Created a new pooled channel, now: 0 active connections, 0 inactive connections and **_5 pending acquire requests._**

But inspite of timeout should occur for some requests, its still live in queue and wait connection

Logs and test are attached

log.txt
ReactorNettyPoolWebfluxTests2.txt

dependencies

io.projectreactor.netty reactor-netty-http 1.0.33 io.projectreactor.netty reactor-netty-core 1.0.33

java version 1.8
OS windows 11

[violetagg]

@manaldush This log:

Created a new pooled channel, now: 0 active connections, 0 inactive connections and **_5 pending acquire requests._**

Is printed at the very beginning when there are no connections in the pool. So it is accurate - no active, no inactive, 5 pending and the connection pool will establish a new connection in order to process the pending requests.

As it is described in the documentation
https://projectreactor.io/docs/netty/release/reference/index.html#_connection_pool_2

pendingAcquireTimeout - The maximum time before which a pending acquire must complete, or a TimeoutException is thrown (resolution: ms). If -1 is specified, no such timeout is applied. Default: 45 seconds.

So this timeout is for the acquire operation, whether the connection pool will provide an already opened connection or it will establish a new connection, depends on the state of the pool.

[manaldush]

@violetagg, thanks for answer, but it is still not clear

06/05/2024 19:06:40.881 [DEBUG] reactor.netty.resources.PooledConnectionProvider:reactor-http-nio-39 [Loggers.java:259:debug] {} > [b7fc1d8f, L:/127.0.0.1:59932 - R:localhost/127.0.0.1:8000] Channel connected, now: 1 active connections, 0 inactive connections and 4 pending acquire requests.
....
06/05/2024 19:06:44.100 [DEBUG] reactor.netty.resources.PooledConnectionProvider:reactor-http-nio-39 [Loggers.java:259:debug] {} > [b7fc1d8f, L:/127.0.0.1:59932 - R:localhost/127.0.0.1:8000] Channel cleaned, now: 0 active connections, 1 inactive connections and 4 pending acquire requests.

In 4 sec I still have requests in pending queue, but pendingAcquireTimeout = 200ms, and I expected that pending acquire operation for pending requests should be timeout at 19:06:41, isn't that?

[violetagg]

@manaldush I agree. There is an issue in Reactor Pool - moving the issue to the proper project.

[violetagg]

Issue can be reproduced with

@ParameterizedTestWithName
@MethodSource("allPools")
void pendingTimeout(PoolStyle configAdjuster) {
	AtomicInteger resource = new AtomicInteger();
	PoolBuilder<AtomicInteger, ?> builder = PoolBuilder
			.from(Mono.just(resource))
			.sizeBetween(0, 1)
			.maxPendingAcquire(10);
	AbstractPool<AtomicInteger> pool = configAdjuster.apply(builder);

	ExecutorService executorService = Executors.newFixedThreadPool(20);
	CompletableFuture<?> f1 = CompletableFuture.runAsync(() -> pool.acquire(Duration.ofMillis(10)).block(), executorService);
	CompletableFuture<?> f2 = CompletableFuture.runAsync(() -> pool.acquire(Duration.ofMillis(10)).block(), executorService);
	CompletableFuture<?> f3 = CompletableFuture.runAsync(() -> pool.acquire(Duration.ofMillis(10)).block(), executorService);
	CompletableFuture<?> f4 = CompletableFuture.runAsync(() -> pool.acquire(Duration.ofMillis(10)).block(), executorService);

	StepVerifier.create(Mono.fromCompletionStage(CompletableFuture.allOf(f1, f2, f3, f4)))
			.expectError()
			.verify(Duration.ofSeconds(5));
}