Auto-configure observability for R2DBC

The new ConnectionFactoryDecorator can be used to decorate the
ConnectionFactory built by the ConnectionFactoryBuilder.

The new R2dbcObservationAutoConfiguration configures a
ConnectionFactoryDecorator to attach a ObservationProxyExecutionListener
to ConnectionFactories. This enables Micrometer Observations for R2DBC
queries.

Closes gh-33768

spring-boot-project/spring-boot-actuator-autoconfigure/build.gradle

@@ -74,6 +74,7 @@ dependencies {
 	optional("io.projectreactor.netty:reactor-netty-http")
 	optional("io.r2dbc:r2dbc-pool")
 	optional("io.r2dbc:r2dbc-spi")
+	optional("io.r2dbc:r2dbc-proxy")
 	optional("jakarta.jms:jakarta.jms-api")
 	optional("jakarta.persistence:jakarta.persistence-api")
 	optional("jakarta.servlet:jakarta.servlet-api")

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/java/org/springframework/boot/actuate/autoconfigure/r2dbc/R2dbcObservationAutoConfiguration.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2012-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.actuate.autoconfigure.r2dbc;
+
+import io.micrometer.observation.ObservationRegistry;
+import io.r2dbc.proxy.ProxyConnectionFactory;
+import io.r2dbc.proxy.observation.ObservationProxyExecutionListener;
+import io.r2dbc.proxy.observation.QueryObservationConvention;
+import io.r2dbc.proxy.observation.QueryParametersTagProvider;
+import io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.ConnectionFactoryOptions;
+
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.boot.actuate.autoconfigure.observation.ObservationAutoConfiguration;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.r2dbc.ConnectionFactoryDecorator;
+import org.springframework.boot.r2dbc.OptionsCapableConnectionFactory;
+import org.springframework.context.annotation.Bean;
+
+/**
+ * {@link EnableAutoConfiguration Auto-configuration} for R2DBC observability support.
+ *
+ * @author Moritz Halbritter
+ * @since 3.2.0
+ */
+@AutoConfiguration(after = ObservationAutoConfiguration.class)
+@ConditionalOnClass({ ConnectionFactory.class, ProxyConnectionFactory.class })
+@EnableConfigurationProperties(R2dbcObservationProperties.class)
+public class R2dbcObservationAutoConfiguration {
+
+	@Bean
+	@ConditionalOnBean(ObservationRegistry.class)
+	ConnectionFactoryDecorator connectionFactoryDecorator(R2dbcObservationProperties properties,
+			ObservationRegistry observationRegistry,
+			ObjectProvider<QueryObservationConvention> queryObservationConvention,
+			ObjectProvider<QueryParametersTagProvider> queryParametersTagProvider) {
+		return (connectionFactory) -> {
+			ObservationProxyExecutionListener listener = new ObservationProxyExecutionListener(observationRegistry,
+					connectionFactory, extractUrl(connectionFactory));
+			listener.setIncludeParameterValues(properties.isIncludeParameterValues());
+			queryObservationConvention.ifAvailable(listener::setQueryObservationConvention);
+			queryParametersTagProvider.ifAvailable(listener::setQueryParametersTagProvider);
+			return ProxyConnectionFactory.builder(connectionFactory).listener(listener).build();
+		};
+	}
+
+	private String extractUrl(ConnectionFactory connectionFactory) {
+		OptionsCapableConnectionFactory optionsCapableConnectionFactory = OptionsCapableConnectionFactory
+			.unwrapFrom(connectionFactory);
+		if (optionsCapableConnectionFactory == null) {
+			return null;
+		}
+		ConnectionFactoryOptions options = optionsCapableConnectionFactory.getOptions();
+		Object host = options.getValue(ConnectionFactoryOptions.HOST);
+		Object port = options.getValue(ConnectionFactoryOptions.PORT);
+		if (host == null || !(port instanceof Integer portAsInt)) {
+			return null;
+		}
+		// See https://github.com/r2dbc/r2dbc-proxy/issues/135
+		return "r2dbc:dummy://%s:%d/".formatted(host, portAsInt);
+	}
+
+}

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/java/org/springframework/boot/actuate/autoconfigure/r2dbc/R2dbcObservationProperties.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2012-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.actuate.autoconfigure.r2dbc;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Configuration properties for R2DBC observability.
+ *
+ * @author Moritz Halbritter
+ * @since 3.2.0
+ */
+@ConfigurationProperties("management.observations.r2dbc")
+public class R2dbcObservationProperties {
+
+	/**
+	 * Whether to tag actual query parameter values.
+	 */
+	private boolean includeParameterValues;
+
+	public boolean isIncludeParameterValues() {
+		return this.includeParameterValues;
+	}
+
+	public void setIncludeParameterValues(boolean includeParameterValues) {
+		this.includeParameterValues = includeParameterValues;
+	}
+
+}

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -90,6 +90,7 @@ org.springframework.boot.actuate.autoconfigure.observation.ObservationAutoConfig
 org.springframework.boot.actuate.autoconfigure.observation.web.servlet.WebMvcObservationAutoConfiguration
 org.springframework.boot.actuate.autoconfigure.quartz.QuartzEndpointAutoConfiguration
 org.springframework.boot.actuate.autoconfigure.r2dbc.ConnectionFactoryHealthContributorAutoConfiguration
+org.springframework.boot.actuate.autoconfigure.r2dbc.R2dbcObservationAutoConfiguration
 org.springframework.boot.actuate.autoconfigure.data.redis.RedisHealthContributorAutoConfiguration
 org.springframework.boot.actuate.autoconfigure.data.redis.RedisReactiveHealthContributorAutoConfiguration
 org.springframework.boot.actuate.autoconfigure.scheduling.ScheduledTasksEndpointAutoConfiguration
@@ -112,4 +113,4 @@ org.springframework.boot.actuate.autoconfigure.web.exchanges.HttpExchangesEndpoi
 org.springframework.boot.actuate.autoconfigure.web.mappings.MappingsEndpointAutoConfiguration
 org.springframework.boot.actuate.autoconfigure.web.reactive.ReactiveManagementContextAutoConfiguration
 org.springframework.boot.actuate.autoconfigure.web.server.ManagementContextAutoConfiguration
-org.springframework.boot.actuate.autoconfigure.web.servlet.ServletManagementContextAutoConfiguration
+org.springframework.boot.actuate.autoconfigure.web.servlet.ServletManagementContextAutoConfiguration

spring-boot-project/spring-boot-actuator-autoconfigure/src/test/java/org/springframework/boot/actuate/autoconfigure/r2dbc/R2dbcObservationAutoConfigurationTests.java

@@ -0,0 +1,131 @@
+/*
+ * Copyright 2012-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.actuate.autoconfigure.r2dbc;
+
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
+
+import io.micrometer.observation.Observation.Context;
+import io.micrometer.observation.ObservationHandler;
+import io.micrometer.observation.ObservationRegistry;
+import io.r2dbc.spi.ConnectionFactory;
+import org.awaitility.Awaitility;
+import org.hamcrest.Matchers;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
+
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.context.annotation.ImportCandidates;
+import org.springframework.boot.r2dbc.ConnectionFactoryBuilder;
+import org.springframework.boot.r2dbc.ConnectionFactoryDecorator;
+import org.springframework.boot.test.context.FilteredClassLoader;
+import org.springframework.boot.test.context.assertj.AssertableApplicationContext;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link R2dbcObservationAutoConfiguration}.
+ *
+ * @author Moritz Halbritter
+ */
+class R2dbcObservationAutoConfigurationTests {
+
+	private final ApplicationContextRunner runnerWithoutObservationRegistry = new ApplicationContextRunner()
+		.withConfiguration(AutoConfigurations.of(R2dbcObservationAutoConfiguration.class));
+
+	private final ApplicationContextRunner runner = this.runnerWithoutObservationRegistry
+		.withBean(ObservationRegistry.class, ObservationRegistry::create);
+
+	@Test
+	void shouldBeRegisteredInAutoConfigurationImports() {
+		assertThat(ImportCandidates.load(AutoConfiguration.class, null).getCandidates())
+			.contains(R2dbcObservationAutoConfiguration.class.getName());
+	}
+
+	@Test
+	void shouldSupplyConnectionFactoryDecorator() {
+		this.runner.run((context) -> assertThat(context).hasSingleBean(ConnectionFactoryDecorator.class));
+	}
+
+	@Test
+	void shouldNotSupplyBeansIfR2dbcSpiIsNotOnClasspath() {
+		this.runner.withClassLoader(new FilteredClassLoader("io.r2dbc.spi"))
+			.run((context) -> assertThat(context).doesNotHaveBean(ConnectionFactoryDecorator.class));
+	}
+
+	@Test
+	void shouldNotSupplyBeansIfR2dbcProxyIsNotOnClasspath() {
+		this.runner.withClassLoader(new FilteredClassLoader("io.r2dbc.proxy"))
+			.run((context) -> assertThat(context).doesNotHaveBean(ConnectionFactoryDecorator.class));
+	}
+
+	@Test
+	void shouldNotSupplyBeansIfObservationRegistryIsNotPresent() {
+		this.runnerWithoutObservationRegistry
+			.run((context) -> assertThat(context).doesNotHaveBean(ConnectionFactoryDecorator.class));
+	}
+
+	@Test
+	void decoratorShouldReportObservations() {
+		this.runner.run((context) -> {
+			CapturingObservationHandler handler = registerCapturingObservationHandler(context);
+			ConnectionFactoryDecorator decorator = context.getBean(ConnectionFactoryDecorator.class);
+			assertThat(decorator).isNotNull();
+			ConnectionFactory connectionFactory = ConnectionFactoryBuilder
+				.withUrl("r2dbc:h2:mem:///" + UUID.randomUUID())
+				.build();
+			ConnectionFactory decorated = decorator.decorate(connectionFactory);
+			Mono.from(decorated.create())
+				.flatMap((c) -> Mono.from(c.createStatement("SELECT 1;").execute())
+					.flatMap((ignore) -> Mono.from(c.close())))
+				.block();
+			assertThat(handler.awaitContext().getName()).as("context.getName()").isEqualTo("r2dbc.query");
+		});
+	}
+
+	private static CapturingObservationHandler registerCapturingObservationHandler(
+			AssertableApplicationContext context) {
+		ObservationRegistry observationRegistry = context.getBean(ObservationRegistry.class);
+		assertThat(observationRegistry).isNotNull();
+		CapturingObservationHandler handler = new CapturingObservationHandler();
+		observationRegistry.observationConfig().observationHandler(handler);
+		return handler;
+	}
+
+	private static class CapturingObservationHandler implements ObservationHandler<Context> {
+
+		private final AtomicReference<Context> context = new AtomicReference<>();
+
+		@Override
+		public boolean supportsContext(Context context) {
+			return true;
+		}
+
+		@Override
+		public void onStart(Context context) {
+			this.context.set(context);
+		}
+
+		Context awaitContext() {
+			return Awaitility.await().untilAtomic(this.context, Matchers.notNullValue());
+		}
+
+	}
+
+}

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/r2dbc/ConnectionFactoryConfigurations.java

@@ -33,6 +33,7 @@ import org.springframework.boot.context.properties.PropertyMapper;
 import org.springframework.boot.context.properties.bind.BindResult;
 import org.springframework.boot.context.properties.bind.Bindable;
 import org.springframework.boot.context.properties.bind.Binder;
+import org.springframework.boot.r2dbc.ConnectionFactoryDecorator;
 import org.springframework.boot.r2dbc.EmbeddedDatabaseConnection;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Condition;
@@ -54,12 +55,14 @@ import org.springframework.util.StringUtils;
  * @author Moritz Halbritter
  * @author Andy Wilkinson
  * @author Phillip Webb
+ * @author Moritz Halbritter
  */
 abstract class ConnectionFactoryConfigurations {
 
 	protected static ConnectionFactory createConnectionFactory(R2dbcProperties properties,
 			R2dbcConnectionDetails connectionDetails, ClassLoader classLoader,
-			List<ConnectionFactoryOptionsBuilderCustomizer> optionsCustomizers) {
+			List<ConnectionFactoryOptionsBuilderCustomizer> optionsCustomizers,
+			List<ConnectionFactoryDecorator> decorators) {
 		try {
 			return org.springframework.boot.r2dbc.ConnectionFactoryBuilder
 				.withOptions(new ConnectionFactoryOptionsInitializer().initialize(properties, connectionDetails,
@@ -69,6 +72,7 @@ abstract class ConnectionFactoryConfigurations {
 						optionsCustomizer.customize(options);
 					}
 				})
+				.decorators(decorators)
 				.build();
 		}
 		catch (IllegalStateException ex) {
@@ -93,10 +97,11 @@ abstract class ConnectionFactoryConfigurations {
 			@Bean(destroyMethod = "dispose")
 			ConnectionPool connectionFactory(R2dbcProperties properties,
 					ObjectProvider<R2dbcConnectionDetails> connectionDetails, ResourceLoader resourceLoader,
-					ObjectProvider<ConnectionFactoryOptionsBuilderCustomizer> customizers) {
+					ObjectProvider<ConnectionFactoryOptionsBuilderCustomizer> customizers,
+					ObjectProvider<ConnectionFactoryDecorator> decorators) {
 				ConnectionFactory connectionFactory = createConnectionFactory(properties,
 						connectionDetails.getIfAvailable(), resourceLoader.getClassLoader(),
-						customizers.orderedStream().toList());
+						customizers.orderedStream().toList(), decorators.orderedStream().toList());
 				R2dbcProperties.Pool pool = properties.getPool();
 				PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
 				ConnectionPoolConfiguration.Builder builder = ConnectionPoolConfiguration.builder(connectionFactory);
@@ -126,9 +131,11 @@ abstract class ConnectionFactoryConfigurations {
 		@Bean
 		ConnectionFactory connectionFactory(R2dbcProperties properties,
 				ObjectProvider<R2dbcConnectionDetails> connectionDetails, ResourceLoader resourceLoader,
-				ObjectProvider<ConnectionFactoryOptionsBuilderCustomizer> customizers) {
+				ObjectProvider<ConnectionFactoryOptionsBuilderCustomizer> customizers,
+				ObjectProvider<ConnectionFactoryDecorator> decorators) {
 			return createConnectionFactory(properties, connectionDetails.getIfAvailable(),
-					resourceLoader.getClassLoader(), customizers.orderedStream().toList());
+					resourceLoader.getClassLoader(), customizers.orderedStream().toList(),
+					decorators.orderedStream().toList());
 		}
 
 	}

spring-boot-project/spring-boot-docs/src/docs/asciidoc/actuator/observability.adoc

@@ -16,10 +16,12 @@ You can additionally register any number of `ObservationRegistryCustomizer` bean
 
 For more details please see the https://micrometer.io/docs/observation[Micrometer Observation documentation].
 
-TIP: Observability for JDBC and R2DBC can be configured using separate projects.
-For JDBC, the https://github.com/jdbc-observations/datasource-micrometer[Datasource Micrometer project] provides a Spring Boot starter which automatically creates observations when JDBC operations are invoked.
+TIP: Observability for JDBC can be configured using a separate project.
+The https://github.com/jdbc-observations/datasource-micrometer[Datasource Micrometer project] provides a Spring Boot starter which automatically creates observations when JDBC operations are invoked.
 Read more about it https://jdbc-observations.github.io/datasource-micrometer/docs/current/docs/html/[in the reference documentation].
-For R2DBC, the https://github.com/spring-projects-experimental/r2dbc-micrometer-spring-boot[Spring Boot Auto Configuration for R2DBC Observation] creates observations for R2DBC query invocations.
+
+TIP: Observability for R2DBC is built into Spring Boot.
+To enable it, add the `io.r2dbc:r2dbc-proxy` dependency to your project.
 
 [[actuator.observability.common-key-values]]
 === Common Key-Values

spring-boot-project/spring-boot/src/main/java/org/springframework/boot/r2dbc/ConnectionFactoryBuilder.java

@@ -17,6 +17,8 @@
 package org.springframework.boot.r2dbc;
 
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Locale;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -43,6 +45,7 @@ import org.springframework.util.ClassUtils;
  * @author Tadaya Tsuyukubo
  * @author Stephane Nicoll
  * @author Andy Wilkinson
+ * @author Moritz Halbritter
  * @since 2.5.0
  */
 public final class ConnectionFactoryBuilder {
@@ -62,6 +65,8 @@ public final class ConnectionFactoryBuilder {
 
 	private final Builder optionsBuilder;
 
+	private final List<ConnectionFactoryDecorator> decorators = new ArrayList<>();
+
 	private ConnectionFactoryBuilder(Builder optionsBuilder) {
 		this.optionsBuilder = optionsBuilder;
 	}
@@ -168,13 +173,41 @@ public final class ConnectionFactoryBuilder {
 		return configure((options) -> options.option(ConnectionFactoryOptions.DATABASE, database));
 	}
 
+	/**
+	 * Add a {@link ConnectionFactoryDecorator decorator}.
+	 * @param decorator the decorator to add
+	 * @return this for method chaining
+	 * @since 3.2.0
+	 */
+	public ConnectionFactoryBuilder decorator(ConnectionFactoryDecorator decorator) {
+		this.decorators.add(decorator);
+		return this;
+	}
+
+	/**
+	 * Add {@link ConnectionFactoryDecorator decorators}.
+	 * @param decorators the decorators to add
+	 * @return this for method chaining
+	 * @since 3.2.0
+	 */
+	public ConnectionFactoryBuilder decorators(Iterable<ConnectionFactoryDecorator> decorators) {
+		for (ConnectionFactoryDecorator decorator : decorators) {
+			this.decorators.add(decorator);
+		}
+		return this;
+	}
+
 	/**
 	 * Build a {@link ConnectionFactory} based on the state of this builder.
 	 * @return a connection factory
 	 */
 	public ConnectionFactory build() {
 		ConnectionFactoryOptions options = buildOptions();
-		return optionsCapableWrapper.buildAndWrap(options);
+		ConnectionFactory connectionFactory = optionsCapableWrapper.buildAndWrap(options);
+		for (ConnectionFactoryDecorator decorator : this.decorators) {
+			connectionFactory = decorator.decorate(connectionFactory);
+		}
+		return connectionFactory;
 	}
 
 	/**

spring-boot-project/spring-boot/src/main/java/org/springframework/boot/r2dbc/ConnectionFactoryDecorator.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2012-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.r2dbc;
+
+import io.r2dbc.spi.ConnectionFactory;
+
+/**
+ * Decorator for {@link ConnectionFactory connection factories}.
+ *
+ * @author Moritz Halbritter
+ * @since 3.2.0
+ * @see ConnectionFactoryBuilder
+ */
+@FunctionalInterface
+public interface ConnectionFactoryDecorator {
+
+	/**
+	 * Decorates the given {@link ConnectionFactory}.
+	 * @param delegate the connection factory which should be decorated
+	 * @return the decorated connection factory
+	 */
+	ConnectionFactory decorate(ConnectionFactory delegate);
+
+}

spring-boot-project/spring-boot/src/test/java/org/springframework/boot/r2dbc/ConnectionFactoryBuilderTests.java

@@ -26,7 +26,9 @@ import io.r2dbc.h2.H2ConnectionFactoryMetadata;
 import io.r2dbc.pool.ConnectionPool;
 import io.r2dbc.pool.ConnectionPoolConfiguration;
 import io.r2dbc.pool.PoolingConnectionFactoryProvider;
+import io.r2dbc.spi.Connection;
 import io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.ConnectionFactoryMetadata;
 import io.r2dbc.spi.ConnectionFactoryOptions;
 import io.r2dbc.spi.Option;
 import io.r2dbc.spi.ValidationDepth;
@@ -34,6 +36,7 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
+import org.reactivestreams.Publisher;
 
 import org.springframework.boot.r2dbc.ConnectionFactoryBuilder.PoolingAwareOptionsCapableWrapper;
 import org.springframework.core.ResolvableType;
@@ -50,6 +53,7 @@ import static org.mockito.Mockito.mock;
  * @author Mark Paluch
  * @author Tadaya Tsuyukubo
  * @author Stephane Nicoll
+ * @author Moritz Halbritter
  */
 class ConnectionFactoryBuilderTests {
 
@@ -235,6 +239,15 @@ class ConnectionFactoryBuilderTests {
 		assertThat(configuration).extracting(expectedOption.property).isEqualTo(expectedOption.value);
 	}
 
+	@Test
+	void shouldApplyDecorators() {
+		String url = "r2dbc:pool:h2:mem:///" + UUID.randomUUID();
+		ConnectionFactory connectionFactory = ConnectionFactoryBuilder.withUrl(url)
+			.decorator((ignored) -> new MyConnectionFactory())
+			.build();
+		assertThat(connectionFactory).isInstanceOf(MyConnectionFactory.class);
+	}
+
 	private static Iterable<Arguments> primitivePoolingConnectionProviderOptions() {
 		return extractPoolingConnectionProviderOptions((field) -> {
 			ResolvableType type = ResolvableType.forField(field);
@@ -320,4 +333,18 @@ class ConnectionFactoryBuilderTests {
 
 	}
 
+	private static class MyConnectionFactory implements ConnectionFactory {
+
+		@Override
+		public Publisher<? extends Connection> create() {
+			return null;
+		}
+
+		@Override
+		public ConnectionFactoryMetadata getMetadata() {
+			return null;
+		}
+
+	}
+
 }