Use native connection factory with message listener containers

This commit updates the auto-configuration to use the native connection
factory for configuring message listener containers. Previously, the
connection factory that could have been wrapped in a caching connection
factory was used.

While using a caching connection factory is suitable for sending
messages (i.e. JmsTemplate usage), it isn't for message listeners as
they need to own the connection for local recovery purposes.

Closes gh-39816

Author: snicoll

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/jms/JmsAnnotationDrivenConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2021 the original author or authors.
+ * Copyright 2012-2024 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.
@@ -25,6 +25,7 @@
 import org.springframework.boot.autoconfigure.condition.ConditionalOnJndi;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnSingleCandidate;
+import org.springframework.boot.jms.ConnectionFactoryUnwrapper;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.jms.annotation.EnableJms;
@@ -89,7 +90,7 @@ DefaultJmsListenerContainerFactoryConfigurer jmsListenerContainerFactoryConfigur
 	DefaultJmsListenerContainerFactory jmsListenerContainerFactory(
 			DefaultJmsListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) {
 		DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
-		configurer.configure(factory, connectionFactory);
+		configurer.configure(factory, ConnectionFactoryUnwrapper.unwrap(connectionFactory));
 		return factory;
 	}
 

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/jms/JmsAutoConfigurationTests.java

@@ -137,7 +137,7 @@ void testDefaultJmsListenerConfiguration() {
 			DefaultMessageListenerContainer container = containerFactory.createListenerContainer(jmsListenerEndpoint);
 			assertThat(container.getClientId()).isNull();
 			assertThat(container.getConcurrentConsumers()).isEqualTo(1);
-			assertThat(container.getConnectionFactory()).isSameAs(connectionFactory);
+			assertThat(container.getConnectionFactory()).isSameAs(connectionFactory.getTargetConnectionFactory());
 			assertThat(container.getMaxConcurrentConsumers()).isEqualTo(1);
 			assertThat(container.getSessionAcknowledgeMode()).isEqualTo(Session.AUTO_ACKNOWLEDGE);
 			assertThat(container.isAutoStartup()).isTrue();

spring-boot-project/spring-boot-docs/src/docs/antora/modules/reference/pages/messaging/jms.adoc

@@ -144,6 +144,10 @@ When the JMS infrastructure is present, any bean can be annotated with `@JmsList
 If no `JmsListenerContainerFactory` has been defined, a default one is configured automatically.
 If a `DestinationResolver`, a `MessageConverter`, or a `jakarta.jms.ExceptionListener` beans are defined, they are associated automatically with the default factory.
 
+In most scenarios, message listener containers should be configured against the native `ConnectionFactory`.
+This way each listener container has its own connection and this gives full responsibility to it in terms of local recovery.
+The auto-configuration uses `ConnectionFactoryUnwrapper` to unwrap the native connection factory from the auto-configured one.
+
 By default, the default factory is transactional.
 If you run in an infrastructure where a `JtaTransactionManager` is present, it is associated to the listener container by default.
 If not, the `sessionTransacted` flag is enabled.
@@ -163,6 +167,8 @@ For instance, the following example exposes another factory that uses a specific
 
 include-code::custom/MyJmsConfiguration[]
 
+NOTE: In the example above, the customization uses `ConnectionFactoryUnwrapper` to associate the native connection factory to the message listener container the same way the auto-configured factory does.
+
 Then you can use the factory in any `@JmsListener`-annotated method as follows:
 
 include-code::custom/MyBean[]

spring-boot-project/spring-boot-docs/src/main/java/org/springframework/boot/docs/howto/messaging/disabletransactedjmssession/MyJmsConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2021 the original author or authors.
+ * Copyright 2012-2024 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.
@@ -19,6 +19,7 @@
 import jakarta.jms.ConnectionFactory;
 
 import org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer;
+import org.springframework.boot.jms.ConnectionFactoryUnwrapper;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
@@ -30,7 +31,7 @@ public class MyJmsConfiguration {
 	public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(ConnectionFactory connectionFactory,
 			DefaultJmsListenerContainerFactoryConfigurer configurer) {
 		DefaultJmsListenerContainerFactory listenerFactory = new DefaultJmsListenerContainerFactory();
-		configurer.configure(listenerFactory, connectionFactory);
+		configurer.configure(listenerFactory, ConnectionFactoryUnwrapper.unwrap(connectionFactory));
 		listenerFactory.setTransactionManager(null);
 		listenerFactory.setSessionTransacted(false);
 		return listenerFactory;

spring-boot-project/spring-boot-docs/src/main/java/org/springframework/boot/docs/messaging/jms/receiving/custom/MyJmsConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2021 the original author or authors.
+ * Copyright 2012-2024 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.
@@ -19,6 +19,7 @@
 import jakarta.jms.ConnectionFactory;
 
 import org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer;
+import org.springframework.boot.jms.ConnectionFactoryUnwrapper;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
@@ -27,16 +28,12 @@
 public class MyJmsConfiguration {
 
 	@Bean
-	public DefaultJmsListenerContainerFactory myFactory(DefaultJmsListenerContainerFactoryConfigurer configurer) {
+	public DefaultJmsListenerContainerFactory myFactory(DefaultJmsListenerContainerFactoryConfigurer configurer,
+			ConnectionFactory connectionFactory) {
 		DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
-		ConnectionFactory connectionFactory = getCustomConnectionFactory();
-		configurer.configure(factory, connectionFactory);
+		configurer.configure(factory, ConnectionFactoryUnwrapper.unwrap(connectionFactory));
 		factory.setMessageConverter(new MyMessageConverter());
 		return factory;
 	}
 
-	private ConnectionFactory getCustomConnectionFactory() {
-		return /**/ null;
-	}
-
 }

spring-boot-project/spring-boot-docs/src/main/kotlin/org/springframework/boot/docs/howto/messaging/disabletransactedjmssession/MyJmsConfiguration.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2022 the original author or authors.
+ * Copyright 2012-2024 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.
@@ -17,6 +17,7 @@
 package org.springframework.boot.docs.howto.messaging.disabletransactedjmssession
 
 import jakarta.jms.ConnectionFactory
+import org.springframework.boot.jms.ConnectionFactoryUnwrapper
 import org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
@@ -30,7 +31,7 @@ class MyJmsConfiguration {
 	fun jmsListenerContainerFactory(connectionFactory: ConnectionFactory?,
 			configurer: DefaultJmsListenerContainerFactoryConfigurer): DefaultJmsListenerContainerFactory {
 		val listenerFactory = DefaultJmsListenerContainerFactory()
-		configurer.configure(listenerFactory, connectionFactory)
+		configurer.configure(listenerFactory, ConnectionFactoryUnwrapper.unwrap(connectionFactory))
 		listenerFactory.setTransactionManager(null)
 		listenerFactory.setSessionTransacted(false)
 		return listenerFactory

spring-boot-project/spring-boot-docs/src/main/kotlin/org/springframework/boot/docs/messaging/jms/receiving/custom/MyJmsConfiguration.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2022 the original author or authors.
+ * Copyright 2012-2024 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.
@@ -18,6 +18,7 @@ package org.springframework.boot.docs.messaging.jms.receiving.custom
 
 import jakarta.jms.ConnectionFactory
 import org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer
+import org.springframework.boot.jms.ConnectionFactoryUnwrapper
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 import org.springframework.jms.config.DefaultJmsListenerContainerFactory
@@ -26,16 +27,12 @@ import org.springframework.jms.config.DefaultJmsListenerContainerFactory
 class MyJmsConfiguration {
 
 	@Bean
-	fun myFactory(configurer: DefaultJmsListenerContainerFactoryConfigurer): DefaultJmsListenerContainerFactory {
+	fun myFactory(configurer: DefaultJmsListenerContainerFactoryConfigurer,
+				  connectionFactory: ConnectionFactory): DefaultJmsListenerContainerFactory {
 		val factory = DefaultJmsListenerContainerFactory()
-		val connectionFactory = getCustomConnectionFactory()
-		configurer.configure(factory, connectionFactory)
+		configurer.configure(factory, ConnectionFactoryUnwrapper.unwrap(connectionFactory))
 		factory.setMessageConverter(MyMessageConverter())
 		return factory
 	}
 
-	fun getCustomConnectionFactory() : ConnectionFactory? {
-		return /**/ null
-	}
-
 }

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

@@ -73,11 +73,15 @@ dependencies {
 	optional("org.liquibase:liquibase-core") {
 		exclude(group: "javax.xml.bind", module: "jaxb-api")
 	}
+	optional("org.messaginghub:pooled-jms") {
+		exclude group: "org.apache.geronimo.specs", module: "geronimo-jms_2.0_spec"
+	}
 	optional("org.postgresql:postgresql")
 	optional("org.slf4j:jul-to-slf4j")
 	optional("org.slf4j:slf4j-api")
 	optional("org.springframework:spring-messaging")
 	optional("org.springframework:spring-orm")
+	optional("org.springframework:spring-jms")
 	optional("org.springframework:spring-oxm")
 	optional("org.springframework:spring-r2dbc")
 	optional("org.springframework:spring-test")

spring-boot-project/spring-boot/src/main/java/org/springframework/boot/jms/ConnectionFactoryUnwrapper.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2012-2024 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.jms;
+
+import jakarta.jms.ConnectionFactory;
+import org.messaginghub.pooled.jms.JmsPoolConnectionFactory;
+
+import org.springframework.jms.connection.CachingConnectionFactory;
+
+/**
+ * Unwrap a {@link ConnectionFactory} that may have been wrapped to perform caching or
+ * pooling.
+ *
+ * @author Stephane Nicoll
+ * @since 6.4.0
+ */
+public abstract class ConnectionFactoryUnwrapper {
+
+	/**
+	 * Return the native {@link ConnectionFactory} by unwrapping it from a cache or pool
+	 * connection factory. Return the given {@link ConnectionFactory} if no caching
+	 * wrapper has been detected.
+	 * @param connectionFactory a connection factory
+	 * @return the native connection factory that it wraps, if any
+	 */
+	public static ConnectionFactory unwrap(ConnectionFactory connectionFactory) {
+		if (connectionFactory instanceof CachingConnectionFactory ccf) {
+			return unwrap(ccf.getTargetConnectionFactory());
+		}
+		ConnectionFactory unwrapedConnectionFactory = unwrapFromJmsPoolConnectionFactory(connectionFactory);
+		return (unwrapedConnectionFactory != null) ? unwrap(unwrapedConnectionFactory) : connectionFactory;
+	}
+
+	private static ConnectionFactory unwrapFromJmsPoolConnectionFactory(ConnectionFactory connectionFactory) {
+		try {
+			if (connectionFactory instanceof JmsPoolConnectionFactory poolConnectionFactory) {
+				return (ConnectionFactory) poolConnectionFactory.getConnectionFactory();
+			}
+		}
+		catch (Throwable ex) {
+			// ignore
+		}
+		return null;
+	}
+
+}

spring-boot-project/spring-boot/src/test/java/org/springframework/boot/jms/ConnectionFactoryUnwrapperTests.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2012-2024 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.jms;
+
+import jakarta.jms.ConnectionFactory;
+import org.junit.jupiter.api.Test;
+import org.messaginghub.pooled.jms.JmsPoolConnectionFactory;
+
+import org.springframework.jms.connection.CachingConnectionFactory;
+import org.springframework.jms.connection.SingleConnectionFactory;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Tests for {@link ConnectionFactoryUnwrapper}.
+ *
+ * @author Stephane Nicoll
+ */
+class ConnectionFactoryUnwrapperTests {
+
+	@Test
+	void unwrapWithSingleConnectionFactory() {
+		ConnectionFactory connectionFactory = new SingleConnectionFactory();
+		assertThat(ConnectionFactoryUnwrapper.unwrap(connectionFactory)).isSameAs(connectionFactory);
+	}
+
+	@Test
+	void unwrapWithConnectionFactory() {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		assertThat(ConnectionFactoryUnwrapper.unwrap(connectionFactory)).isSameAs(connectionFactory);
+	}
+
+	@Test
+	void unwrapWithCachingConnectionFactory() {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		assertThat(ConnectionFactoryUnwrapper.unwrap(new CachingConnectionFactory(connectionFactory)))
+			.isSameAs(connectionFactory);
+	}
+
+	@Test
+	void unwrapWithNestedCachingConnectionFactories() {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		CachingConnectionFactory firstCachingConnectionFactory = new CachingConnectionFactory(connectionFactory);
+		CachingConnectionFactory secondCachingConnectionFactory = new CachingConnectionFactory(
+				firstCachingConnectionFactory);
+		assertThat(ConnectionFactoryUnwrapper.unwrap(secondCachingConnectionFactory)).isSameAs(connectionFactory);
+	}
+
+	@Test
+	void unwrapWithJmsPoolConnectionFactory() {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		JmsPoolConnectionFactory poolConnectionFactory = new JmsPoolConnectionFactory();
+		poolConnectionFactory.setConnectionFactory(connectionFactory);
+		assertThat(ConnectionFactoryUnwrapper.unwrap(poolConnectionFactory)).isSameAs(connectionFactory);
+	}
+
+	@Test
+	void unwrapWithNestedJmsPoolConnectionFactories() {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		JmsPoolConnectionFactory firstPooledConnectionFactory = new JmsPoolConnectionFactory();
+		firstPooledConnectionFactory.setConnectionFactory(connectionFactory);
+		JmsPoolConnectionFactory secondPooledConnectionFactory = new JmsPoolConnectionFactory();
+		secondPooledConnectionFactory.setConnectionFactory(firstPooledConnectionFactory);
+		assertThat(ConnectionFactoryUnwrapper.unwrap(secondPooledConnectionFactory)).isSameAs(connectionFactory);
+	}
+
+}