When an explicit EventLoopGroup is configured ensure only one connection pool is created (#3321)

Fixes #3316

Author: violetagg

reactor-netty-core/src/main/java/reactor/netty/transport/Transport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020-2021 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2020-2024 VMware, Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -231,8 +231,7 @@ public <O> T option(ChannelOption<O> key, @Nullable O value) {
 	 * @return a new {@link Transport} reference
 	 */
 	public T runOn(EventLoopGroup eventLoopGroup) {
-		Objects.requireNonNull(eventLoopGroup, "eventLoopGroup");
-		return runOn(preferNative -> eventLoopGroup);
+		return runOn(new EventLoopGroupLoopResources(eventLoopGroup));
 	}
 
 	/**
@@ -389,4 +388,38 @@ public final T wiretap(String category, LogLevel level, AdvancedByteBufFormat fo
 	protected abstract T duplicate();
 
 	static final Logger log = Loggers.getLogger(Transport.class);
+
+	static final class EventLoopGroupLoopResources implements LoopResources {
+
+		final EventLoopGroup eventLoopGroup;
+
+		EventLoopGroupLoopResources(EventLoopGroup eventLoopGroup) {
+			Objects.requireNonNull(eventLoopGroup, "eventLoopGroup");
+			this.eventLoopGroup = eventLoopGroup;
+		}
+
+		@Override
+		public EventLoopGroup onServer(boolean useNative) {
+			return eventLoopGroup;
+		}
+
+		@Override
+		public boolean equals(Object o) {
+			if (this == o) {
+				return true;
+			}
+			if (o == null || getClass() != o.getClass()) {
+				return false;
+			}
+			EventLoopGroupLoopResources that = (EventLoopGroupLoopResources) o;
+			return Objects.equals(eventLoopGroup, that.eventLoopGroup);
+		}
+
+		@Override
+		public int hashCode() {
+			int result = 1;
+			result = 31 * result + Objects.hashCode(eventLoopGroup);
+			return result;
+		}
+	}
 }

reactor-netty-core/src/test/java/reactor/netty/resources/DefaultPooledConnectionProviderTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2017-2021 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2017-2024 VMware, Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -25,6 +25,7 @@
 import java.util.Map;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
@@ -51,6 +52,7 @@
 import reactor.core.publisher.Signal;
 import reactor.netty.Connection;
 import reactor.netty.ConnectionObserver;
+import reactor.netty.DisposableChannel;
 import reactor.netty.DisposableServer;
 import reactor.netty.SocketUtils;
 import reactor.netty.channel.ChannelMetricsRecorder;
@@ -527,6 +529,38 @@ private void doTestIssue1790(boolean fifoPool) {
 		}
 	}
 
+	@Test
+	void testIssue3316() throws ExecutionException, InterruptedException {
+		DisposableServer disposableServer =
+				TcpServer.create()
+				         .port(0)
+				         .handle((in, out) -> out.send(in.receive().retain()))
+				         .bindNow();
+
+		DefaultPooledConnectionProvider provider =
+				(DefaultPooledConnectionProvider) ConnectionProvider.create("testIssue3316", 400);
+		EventLoopGroup group = new NioEventLoopGroup();
+		try {
+			Flux.range(0, 400)
+			    .flatMap(i ->
+			        TcpClient.create(provider)
+			                 .port(disposableServer.port())
+			                 .runOn(group)
+			                 .connect())
+			    .doOnNext(DisposableChannel::dispose)
+			    .blockLast(Duration.ofSeconds(5));
+
+			assertThat(provider.channelPools.size()).isEqualTo(1);
+		}
+		finally {
+			disposableServer.disposeNow();
+			provider.disposeLater()
+			        .block(Duration.ofSeconds(5));
+			group.shutdownGracefully()
+			     .get();
+		}
+	}
+
 	static final class PoolImpl extends AtomicInteger implements InstrumentedPool<PooledConnection> {
 
 		@Override