Ensure all ConnectionProvider built-in metrics are disabled when a custom recorder is provided

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

@@ -135,7 +135,7 @@ public final Mono<? extends Connection> acquire(
 				boolean metricsEnabled = poolFactory.metricsEnabled || config.metricsRecorder() != null;
 				String id = metricsEnabled ? poolKey.hashCode() + "" : null;
 
-				InstrumentedPool<T> newPool = metricsEnabled && Metrics.isMicrometerAvailable() ?
+				InstrumentedPool<T> newPool = metricsEnabled && poolFactory.registrar == null && Metrics.isMicrometerAvailable() ?
 						createPool(id, config, poolFactory, remoteAddress, resolverGroup) :
 						createPool(config, poolFactory, remoteAddress, resolverGroup);
 

reactor-netty-http/src/test/java/reactor/netty/resources/PooledConnectionProviderDefaultMetricsTest.java

@@ -19,6 +19,8 @@
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.Metrics;
 import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
 import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
 import io.netty.handler.ssl.util.SelfSignedCertificate;
 import org.junit.jupiter.api.AfterEach;
@@ -30,10 +32,12 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.netty.BaseHttpTest;
+import reactor.netty.channel.ChannelMetricsRecorder;
 import reactor.netty.http.Http2SslContextSpec;
 import reactor.netty.http.HttpProtocol;
 import reactor.netty.http.client.HttpClient;
 import reactor.netty.http.server.HttpServer;
+import reactor.test.StepVerifier;
 
 import java.security.cert.CertificateException;
 import java.time.Duration;
@@ -46,6 +50,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
+import static org.mockito.Mockito.mock;
 import static reactor.netty.Metrics.ACTIVE_CONNECTIONS;
 import static reactor.netty.Metrics.ACTIVE_STREAMS;
 import static reactor.netty.Metrics.ERROR;
@@ -56,6 +61,7 @@
 import static reactor.netty.Metrics.MAX_PENDING_CONNECTIONS;
 import static reactor.netty.Metrics.NAME;
 import static reactor.netty.Metrics.PENDING_STREAMS;
+import static reactor.netty.Metrics.REGISTRY;
 import static reactor.netty.Metrics.STATUS;
 import static reactor.netty.Metrics.TOTAL_CONNECTIONS;
 import static reactor.netty.http.client.HttpClientState.STREAM_CONFIGURED;
@@ -376,6 +382,52 @@ private void testIssue3060(ConnectionProvider provider) throws Exception {
 		}
 	}
 
+	/* https://github.com/reactor/reactor-netty/issues/3519 */
+	@Test
+	public void testConnectionProviderDisableAllBuiltInMetrics() throws Exception {
+		disposableServer =
+				createServer()
+				        .handle((req, res) -> res.sendString(Mono.just("testConnectionProviderDisableAllBuiltInMetrics")))
+				        .bindNow();
+
+		ConnectionProvider provider =
+				ConnectionProvider.builder("testConnectionProviderDisableAllBuiltInMetrics")
+				                  .metrics(true, () -> mock(ConnectionProvider.MeterRegistrar.class))
+				                  .build();
+
+		try {
+			CountDownLatch latch = new CountDownLatch(1);
+			createClient(provider, disposableServer.port())
+			        .metrics(true, () -> mock(ChannelMetricsRecorder.class))
+			        .headers(h -> h.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE))
+			        .doOnRequest((req, conn) -> conn.channel().closeFuture().addListener(f -> latch.countDown()))
+			        .get()
+			        .uri("/")
+			        .responseContent()
+			        .aggregate()
+			        .asString()
+			        .as(StepVerifier::create)
+			        .expectNext("testConnectionProviderDisableAllBuiltInMetrics")
+			        .expectComplete()
+			        .verify(Duration.ofSeconds(5));
+
+			assertThat(latch.await(5, TimeUnit.SECONDS)).as("latch await").isTrue();
+
+			AtomicInteger count = new AtomicInteger();
+			REGISTRY.forEachMeter(meter -> {
+				if (meter.getId().getName().startsWith("reactor.netty.connection")) {
+					count.incrementAndGet();
+				}
+			});
+
+			assertThat(count).hasValue(0);
+		}
+		finally {
+			provider.disposeLater()
+			        .block(Duration.ofSeconds(5));
+		}
+	}
+
 	private double getGaugeValue(String gaugeName, String poolName) {
 		Gauge gauge = registry.find(gaugeName).tag(NAME, poolName).gauge();
 		double result = -1;