Cope with hanging call to connection.start() in JmsHealthIndicator

Previously, if the call to connection.start() hung, JmsHealthIndicator
would also hang and then never respond.

This commit introduces the use of an additional thread that waits for
up to 5 seconds for the connection to start. If the call to start
does not complete within that time, the connection is closed. The
call to close causes the call to start to throw an exception, thereby
stopping the hang and allowing the indicator to report that the
broker is down.

Closes gh-10809

spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/jms/JmsHealthIndicator.java

@@ -16,8 +16,15 @@
 
 package org.springframework.boot.actuate.jms;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
 import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 
 import org.springframework.boot.actuate.health.AbstractHealthIndicator;
 import org.springframework.boot.actuate.health.Health;
@@ -31,6 +38,8 @@ import org.springframework.boot.actuate.health.HealthIndicator;
  */
 public class JmsHealthIndicator extends AbstractHealthIndicator {
 
+	private final Log logger = LogFactory.getLog(JmsHealthIndicator.class);
+
 	private final ConnectionFactory connectionFactory;
 
 	public JmsHealthIndicator(ConnectionFactory connectionFactory) {
@@ -41,10 +50,48 @@ public class JmsHealthIndicator extends AbstractHealthIndicator {
 	@Override
 	protected void doHealthCheck(Health.Builder builder) throws Exception {
 		try (Connection connection = this.connectionFactory.createConnection()) {
-			connection.start();
+			new MonitoredConnection(connection).start();
 			builder.up().withDetail("provider",
 					connection.getMetaData().getJMSProviderName());
 		}
 	}
 
+	private final class MonitoredConnection {
+
+		private final CountDownLatch latch = new CountDownLatch(1);
+
+		private final Connection connection;
+
+		MonitoredConnection(Connection connection) {
+			this.connection = connection;
+		}
+
+		public void start() throws JMSException {
+			new Thread(() -> {
+				try {
+					if (!this.latch.await(5, TimeUnit.SECONDS)) {
+						JmsHealthIndicator.this.logger.warn(
+								"Connection failed to start within 5 seconds and will be closed.");
+						closeConnection();
+					}
+				}
+				catch (InterruptedException ex) {
+					Thread.currentThread().interrupt();
+				}
+			}, "jms-health-indicator").start();
+			this.connection.start();
+			this.latch.countDown();
+		}
+
+		private void closeConnection() {
+			try {
+				this.connection.close();
+			}
+			catch (Exception ex) {
+				// Continue
+			}
+		}
+
+	}
+
 }

spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/jms/JmsHealthIndicatorTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2017 the original author or authors.
+ * Copyright 2012-2018 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.
@@ -22,6 +22,8 @@ import javax.jms.ConnectionMetaData;
 import javax.jms.JMSException;
 
 import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 
 import org.springframework.boot.actuate.health.Health;
 import org.springframework.boot.actuate.health.Status;
@@ -29,6 +31,7 @@ import org.springframework.boot.actuate.health.Status;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willThrow;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -97,4 +100,49 @@ public class JmsHealthIndicatorTests {
 		assertThat(health.getDetails().get("provider")).isNull();
 	}
 
+	@Test
+	public void whenConnectionStartIsUnresponsiveStatusIsDown() throws JMSException {
+		ConnectionMetaData connectionMetaData = mock(ConnectionMetaData.class);
+		given(connectionMetaData.getJMSProviderName()).willReturn("JMS test provider");
+		Connection connection = mock(Connection.class);
+		UnresponsiveStartAnswer unresponsiveStartAnswer = new UnresponsiveStartAnswer();
+		doAnswer(unresponsiveStartAnswer).when(connection).start();
+		doAnswer((invocation) -> {
+			unresponsiveStartAnswer.connectionClosed();
+			return null;
+		}).when(connection).close();
+		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);
+		assertThat((String) health.getDetails().get("error"))
+				.contains("Connection closed");
+	}
+
+	private static final class UnresponsiveStartAnswer implements Answer<Void> {
+
+		private boolean connectionClosed = false;
+
+		private final Object monitor = new Object();
+
+		@Override
+		public Void answer(InvocationOnMock invocation) throws Throwable {
+			synchronized (this.monitor) {
+				while (!this.connectionClosed) {
+					this.monitor.wait();
+				}
+			}
+			throw new JMSException("Connection closed");
+		}
+
+		private void connectionClosed() {
+			synchronized (this.monitor) {
+				this.connectionClosed = true;
+				this.monitor.notifyAll();
+			}
+		}
+
+	}
+
 }