INT-3066 Fix NPE During Connection Recovery

The JmsOutboundGateway's Listener container could throw
an NPE during connection recovery - while logging recovery
status.

The log attempts to include the destination description which
throws the NPE if there is no destination yet established.

Only attempt to get the description if the destination has
been established. And, even then, use a try/catch block
to prevent the NPE. Log the destination as 'null' in this
case.

Add a test case to detect the NPE by adding an ErrorHandlingTaskExecutor.

No errors should be detected with the fix in place.

spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java

@@ -1067,8 +1067,19 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
 			if (this.replyDestination instanceof TemporaryQueue) {
 				return "Temporary queue:" + this.replyDestination.toString();
 			}
+			else if (super.getDestination() != null){
+				try {
+					return super.getDestinationDescription();
+				}
+				catch (Exception e) {
+					if (logger.isWarnEnabled()) {
+						logger.warn("Unexpected error obtaining destination description: " + e.getMessage());
+					}
+					return null;
+				}
+			}
 			else {
-				return super.getDestinationDescription();
+				return null;
 			}
 		}
 
@@ -1081,9 +1092,9 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
 
 	private class TimedReply {
 
-		private long timeStamp = System.currentTimeMillis();
+		private final long timeStamp = System.currentTimeMillis();
 
-		private javax.jms.Message reply;
+		private final javax.jms.Message reply;
 
 		public TimedReply(javax.jms.Message reply) {
 			this.reply = reply;

spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java

@@ -16,16 +16,42 @@
 package org.springframework.integration.jms;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.Session;
+import javax.jms.TemporaryQueue;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import org.springframework.beans.factory.support.DefaultListableBeanFactory;
 import org.springframework.integration.jms.JmsOutboundGateway.ReplyContainerProperties;
 import org.springframework.integration.test.util.TestUtils;
+import org.springframework.integration.util.ErrorHandlingTaskExecutor;
+import org.springframework.jms.JmsException;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.util.ErrorHandler;
 import org.springframework.util.ObjectUtils;
 
-import org.junit.Test;
-
 /**
  * @author Gary Russell
  * @since 2.2.4
@@ -33,6 +59,8 @@ import org.junit.Test;
  */
 public class JmsOutboundGatewayTests {
 
+	final Log logger = LogFactory.getLog(this.getClass());
+
 	@Test
 	public void testContainerBeanNameWhenNoGatewayBeanName() {
 		JmsOutboundGateway gateway = new JmsOutboundGateway();
@@ -45,4 +73,75 @@ public class JmsOutboundGatewayTests {
 				".replyListener",
 			TestUtils.getPropertyValue(gateway, "replyContainer.beanName"));
 	}
+
+	@Test
+	public void testReplyContainerRecovery() throws Exception {
+		JmsOutboundGateway gateway = new JmsOutboundGateway();
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		gateway.setConnectionFactory(connectionFactory);
+		gateway.setRequestDestinationName("foo");
+		gateway.setUseReplyContainer(true);
+		ReplyContainerProperties replyContainerProperties = new ReplyContainerProperties();
+		final List<Throwable> errors = new ArrayList<Throwable>();
+		ErrorHandlingTaskExecutor errorHandlingTaskExecutor = new ErrorHandlingTaskExecutor(Executors.newFixedThreadPool(10), new ErrorHandler() {
+
+			@Override
+			public void handleError(Throwable t) {
+				logger.info("Error:", t);
+				errors.add(t);
+				throw new RuntimeException(t);
+			}
+		});
+		replyContainerProperties.setTaskExecutor(errorHandlingTaskExecutor);
+		replyContainerProperties.setRecoveryInterval(100L);
+		gateway.setReplyContainerProperties(replyContainerProperties);
+		final Connection connection = mock(Connection.class);
+		final AtomicInteger connectionAttempts = new AtomicInteger();
+		doAnswer(new Answer<Connection>() {
+
+			@SuppressWarnings("serial")
+			@Override
+			public Connection answer(InvocationOnMock invocation) throws Throwable {
+				int theCount = connectionAttempts.incrementAndGet();
+				if (theCount > 1 && theCount < 4) {
+					throw new JmsException("bar") {};
+				}
+				return connection;
+			}
+		}).when(connectionFactory).createConnection();
+		Session session = mock(Session.class);
+		when(connection.createSession(false, 1)).thenReturn(session);
+		MessageConsumer consumer = mock(MessageConsumer.class);
+		when(session.createConsumer(any(Destination.class), anyString())).thenReturn(consumer);
+		when(session.createTemporaryQueue()).thenReturn(mock(TemporaryQueue.class));
+		final Message message = mock(Message.class);
+		final AtomicInteger count = new AtomicInteger();
+		doAnswer(new Answer<Message>() {
+
+			@SuppressWarnings("serial")
+			@Override
+			public Message answer(InvocationOnMock invocation) throws Throwable {
+				int theCount = count.incrementAndGet();
+				if (theCount > 1 && theCount < 4) {
+					throw new JmsException("foo") {};
+				}
+				if (theCount > 4) {
+					Thread.sleep(100);
+					return null;
+				}
+				return message;
+			}
+		}).when(consumer).receive(anyLong());
+		when(message.getJMSCorrelationID()).thenReturn("foo");
+		DefaultListableBeanFactory beanFactory = new DefaultListableBeanFactory();
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.initialize();
+		beanFactory.registerSingleton("taskScheduler", taskScheduler);
+		gateway.setBeanFactory(beanFactory);
+		gateway.afterPropertiesSet();
+		gateway.start();
+		Thread.sleep(1000);
+		assertTrue(count.get() > 4);
+		assertEquals(0, errors.size());
+	}
 }