The sequence for acquiring the connection was not cancelled after the request ended.

[qnnn]

Expected Behavior

The sequence for acquiring the connection will be cancelled after the request ends.

Actual Behavior

The sequence for acquiring the connection was not cancelled after the request ended.

Steps to Reproduce

I recently encountered a small issue while using Spring Cloud Gateway. After the request reaches the responseTimeout value and the response is sent back (due to a connection timeout), the program continues to retry connecting to other DNS-resolved addresses. (For example, responseTimeout = 8s, connectionTimeout = 5s, and there are three addresses in the DNS resolution result.)

It seems that after the outer sequence is cancelled, the inner sequence responsible for acquiring the connection is not able to detect this cancellation. Does this behavior align with the phenomenon simulated in the following flow? (I’m just trying to learn.)

public static void main(String[] args) throws InterruptedException {
	execute();
	Thread.sleep(20000);
}

public static void execute() {
	// Similar to PooledConnectionAllocator#connectChannel
	Mono.<String>create(sink -> acquireConnection(sink))
			// Simulate request
			.flatMap(connection -> Mono.just("send request and receive response"))
			// Set response timeout to 8 seconds
			.timeout(Duration.ofSeconds(8), Mono.just("read timeout!"))
			.subscribe(System.out::println);
}

public static void acquireConnection(MonoSink<String> sink) {
	// Similar to TransportConnector#connect
	Mono.just("acquire connection")
			// Set connection timeout to 5 seconds
			.delayElement(Duration.ofSeconds(5))
			.doOnNext(connection -> {
				System.out.println("connection timeout!");
				throw new RuntimeException();
			})
			.onErrorResume(Exception.class, e ->
					Mono.defer(() -> Mono.just("acquire connection")
									.delayElement(Duration.ofSeconds(5))
									.doOnNext(connection -> {
										System.out.println("connection timeout!");
										throw new RuntimeException();
									})
							)
							// Retry connecting other DNS-resolved addresses.
							.retryWhen(RetrySpec.max(2))
			)
			.subscribe(connection -> sink.success(connection));
}

Possible Solution

Your Environment

• Reactor version(s) used: 1.1.20
• JVM version: 17

[qnnn]

Perhaps we could register a cancel callback using sink.onCancel(Subscription::cancel) during onSubscribe?

[violetagg]

@qnnn I'll be working on this. It may need fixes in reactor pool.

[violetagg]

Perhaps we could register a cancel callback using sink.onCancel(Subscription::cancel) during onSubscribe?

We do register cancel callback, see

reactor-netty/reactor-netty-core/src/main/java/reactor/netty/resources/DefaultPooledConnectionProvider.java

Lines 211 to 220 in 40a880e

	public void onSubscribe(Subscription s) {
	if (Operators.validate(subscription, s)) {
	this.subscription = s;
	cancellations.add(this);
	if (!retried) {
	sink.onCancel(cancellations);
	}
	s.request(Long.MAX_VALUE);
	}
	}

[qnnn]

Perhaps we could register a cancel callback using sink.onCancel(Subscription::cancel) during onSubscribe?

We do register cancel callback, see

reactor-netty/reactor-netty-core/src/main/java/reactor/netty/resources/DefaultPooledConnectionProvider.java

Lines 211 to 220 in 40a880e

	public void onSubscribe(Subscription s) {
	if (Operators.validate(subscription, s)) {
	this.subscription = s;
	cancellations.add(this);
	if (!retried) {
	sink.onCancel(cancellations);
	}
	s.request(Long.MAX_VALUE);
	}
	}

@violetagg Thank you for your reply. I have also noticed the reactor-pool project and have attempted to submit a PR #256. Apart from the callback mentioned in your reply, do we need to register a callback for PooledConnectionInitializer and ClientTransportSubscriber as well? In the version I tried to modify, it seems that if callbacks are not registered for them, acquiring a connection cannot be canceled either.

[violetagg]

reactor/reactor-pool#263
reactor/reactor-pool#265

[violetagg]

@qnnn I'm closing this issue. Once an acquisition is started we do not want to stop the connection establishment, so that the connection (if established) will stay in the pool and reused.