From 3154173d23e32ea0f0e8cf3bbf4854c5d0068b54 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 5 Jul 2013 15:38:33 -0400 Subject: [PATCH] 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. --- .../integration/jms/JmsOutboundGateway.java | 17 ++- .../jms/JmsOutboundGatewayTests.java | 103 +++++++++++++++++- 2 files changed, 115 insertions(+), 5 deletions(-) diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java index ba4f243104..9dd7cdb4ba 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java +++ b/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; diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java index c8099582a9..079c15d756 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java +++ b/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 errors = new ArrayList(); + 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() { + + @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() { + + @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()); + } }