Merge branch '4.0.x' into 4.1.x

Closes gh-51433

module/spring-boot-jms/src/main/java/org/springframework/boot/jms/health/JmsHealthIndicator.java

@@ -65,7 +65,7 @@ public class JmsHealthIndicator extends AbstractHealthIndicator {
 		}
 
 		void start() throws JMSException {
-			new Thread(() -> {
+			Thread watchdog = new Thread(() -> {
 				try {
 					if (!this.latch.await(5, TimeUnit.SECONDS)) {
 						JmsHealthIndicator.this.logger
@@ -76,9 +76,15 @@ public class JmsHealthIndicator extends AbstractHealthIndicator {
 				catch (InterruptedException ex) {
 					Thread.currentThread().interrupt();
 				}
-			}, "jms-health-indicator").start();
-			this.connection.start();
-			this.latch.countDown();
+			}, "jms-health-indicator");
+			watchdog.setDaemon(true);
+			watchdog.start();
+			try {
+				this.connection.start();
+			}
+			finally {
+				this.latch.countDown();
+			}
 		}
 
 		private void closeConnection() {

module/spring-boot-jms/src/test/java/org/springframework/boot/jms/health/JmsHealthIndicatorTests.java

@@ -16,6 +16,8 @@
 
 package org.springframework.boot.jms.health;
 
+import java.time.Duration;
+
 import jakarta.jms.Connection;
 import jakarta.jms.ConnectionFactory;
 import jakarta.jms.ConnectionMetaData;
@@ -32,6 +34,7 @@ import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.then;
 import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.BDDMockito.willThrow;
+import static org.mockito.Mockito.after;
 import static org.mockito.Mockito.mock;
 
 /**
@@ -96,6 +99,18 @@ class JmsHealthIndicatorTests {
 		assertThat(health.getDetails()).doesNotContainKey("provider");
 	}
 
+	@Test
+	void whenConnectionStartThrowsWatchdogThreadDoesNotAlsoCloseConnection() throws JMSException {
+		Connection connection = mock(Connection.class);
+		willThrow(new JMSException("Could not start", "123")).given(connection).start();
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		given(connectionFactory.createConnection()).willReturn(connection);
+		JmsHealthIndicator indicator = new JmsHealthIndicator(connectionFactory);
+		Health health = indicator.health();
+		assertThat(health.getStatus()).isEqualTo(Status.DOWN);
+		then(connection).should(after(Duration.ofSeconds(5).plusMillis(500).toMillis()).times(1)).close();
+	}
+
 	@Test
 	void whenConnectionStartIsUnresponsiveStatusIsDown() throws JMSException {
 		ConnectionMetaData connectionMetaData = mock(ConnectionMetaData.class);