Add support for configuring JMS listener sessionTransacted

This commit introduces `spring.jms.listener.session-transacted` property
in order to enable explicit configuration of `sessionTransacted` on the
`DefaultMessageListenerContainer`.

Prior to this commit, `sessionTransacted` would be configured implicitly
based on presence of `JtaTransactionManager`.

Closes gh-37472

Author: vpavic

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2021 the original author or authors.
+ * 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.
@@ -32,6 +32,7 @@
  *
  * @author Stephane Nicoll
  * @author Eddú Meléndez
+ * @author Vedran Pavic
  * @since 1.3.3
  */
 public final class DefaultJmsListenerContainerFactoryConfigurer {
@@ -101,12 +102,16 @@ public void configure(DefaultJmsListenerContainerFactory factory, ConnectionFact
 		Assert.notNull(connectionFactory, "ConnectionFactory must not be null");
 		factory.setConnectionFactory(connectionFactory);
 		factory.setPubSubDomain(this.jmsProperties.isPubSubDomain());
+		JmsProperties.Listener listener = this.jmsProperties.getListener();
 		if (this.transactionManager != null) {
 			factory.setTransactionManager(this.transactionManager);
 		}
-		else {
+		else if (listener.getSessionTransacted() == null) {
 			factory.setSessionTransacted(true);
 		}
+		if (listener.getSessionTransacted() != null) {
+			factory.setSessionTransacted(listener.getSessionTransacted());
+		}
 		if (this.destinationResolver != null) {
 			factory.setDestinationResolver(this.destinationResolver);
 		}
@@ -116,7 +121,6 @@ public void configure(DefaultJmsListenerContainerFactory factory, ConnectionFact
 		if (this.exceptionListener != null) {
 			factory.setExceptionListener(this.exceptionListener);
 		}
-		JmsProperties.Listener listener = this.jmsProperties.getListener();
 		factory.setAutoStartup(listener.isAutoStartup());
 		if (listener.getAcknowledgeMode() != null) {
 			factory.setSessionAcknowledgeMode(listener.getAcknowledgeMode().getMode());

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

@@ -26,6 +26,7 @@
  * @author Greg Turnquist
  * @author Phillip Webb
  * @author Stephane Nicoll
+ * @author Vedran Pavic
  * @since 1.0.0
  */
 @ConfigurationProperties(prefix = "spring.jms")
@@ -145,6 +146,11 @@ public static class Listener {
 		 */
 		private AcknowledgeMode acknowledgeMode;
 
+		/**
+		 * Whether the container should use transacted JMS sessions.
+		 */
+		private Boolean sessionTransacted;
+
 		/**
 		 * Minimum number of concurrent consumers.
 		 */
@@ -178,6 +184,14 @@ public void setAcknowledgeMode(AcknowledgeMode acknowledgeMode) {
 			this.acknowledgeMode = acknowledgeMode;
 		}
 
+		public Boolean getSessionTransacted() {
+			return this.sessionTransacted;
+		}
+
+		public void setSessionTransacted(Boolean sessionTransacted) {
+			this.sessionTransacted = sessionTransacted;
+		}
+
 		public Integer getConcurrency() {
 			return this.concurrency;
 		}

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

@@ -57,6 +57,7 @@
  * @author Stephane Nicoll
  * @author Aurélien Leboulanger
  * @author Eddú Meléndez
+ * @author Vedran Pavic
  */
 class JmsAutoConfigurationTests {
 
@@ -143,15 +144,16 @@ void jmsListenerContainerFactoryWhenMultipleConnectionFactoryBeansShouldBackOff(
 	void testJmsListenerContainerFactoryWithCustomSettings() {
 		this.contextRunner.withUserConfiguration(EnableJmsConfiguration.class)
 			.withPropertyValues("spring.jms.listener.autoStartup=false", "spring.jms.listener.acknowledgeMode=client",
-					"spring.jms.listener.concurrency=2", "spring.jms.listener.receiveTimeout=2s",
-					"spring.jms.listener.maxConcurrency=10")
+					"spring.jms.listener.sessionTransacted=false", "spring.jms.listener.concurrency=2",
+					"spring.jms.listener.receiveTimeout=2s", "spring.jms.listener.maxConcurrency=10")
 			.run(this::testJmsListenerContainerFactoryWithCustomSettings);
 	}
 
 	private void testJmsListenerContainerFactoryWithCustomSettings(AssertableApplicationContext loaded) {
 		DefaultMessageListenerContainer container = getContainer(loaded, "jmsListenerContainerFactory");
 		assertThat(container.isAutoStartup()).isFalse();
 		assertThat(container.getSessionAcknowledgeMode()).isEqualTo(Session.CLIENT_ACKNOWLEDGE);
+		assertThat(container.isSessionTransacted()).isFalse();
 		assertThat(container.getConcurrentConsumers()).isEqualTo(2);
 		assertThat(container.getMaxConcurrentConsumers()).isEqualTo(10);
 		assertThat(container).hasFieldOrPropertyWithValue("receiveTimeout", 2000L);
@@ -179,6 +181,18 @@ void testDefaultContainerFactoryWithJtaTransactionManager() {
 			});
 	}
 
+	@Test
+	void testDefaultContainerFactoryWithJtaTransactionManagerAndSessionTransactedEnabled() {
+		this.contextRunner.withUserConfiguration(TestConfiguration7.class, EnableJmsConfiguration.class)
+				.withPropertyValues("spring.jms.listener.sessionTransacted=true")
+				.run((context) -> {
+					DefaultMessageListenerContainer container = getContainer(context, "jmsListenerContainerFactory");
+					assertThat(container.isSessionTransacted()).isTrue();
+					assertThat(container).hasFieldOrPropertyWithValue("transactionManager",
+							context.getBean(JtaTransactionManager.class));
+				});
+	}
+
 	@Test
 	void testDefaultContainerFactoryNonJtaTransactionManager() {
 		this.contextRunner.withUserConfiguration(TestConfiguration8.class, EnableJmsConfiguration.class)
@@ -198,6 +212,17 @@ void testDefaultContainerFactoryNoTransactionManager() {
 		});
 	}
 
+	@Test
+	void testDefaultContainerFactoryNoTransactionManagerAndSessionTransactedDisabled() {
+		this.contextRunner.withUserConfiguration(EnableJmsConfiguration.class)
+				.withPropertyValues("spring.jms.listener.sessionTransacted=false")
+				.run((context) -> {
+					DefaultMessageListenerContainer container = getContainer(context, "jmsListenerContainerFactory");
+					assertThat(container.isSessionTransacted()).isFalse();
+					assertThat(container).hasFieldOrPropertyWithValue("transactionManager", null);
+				});
+	}
+
 	@Test
 	void testDefaultContainerFactoryWithMessageConverters() {
 		this.contextRunner.withUserConfiguration(MessageConvertersConfiguration.class, EnableJmsConfiguration.class)