QPID-5050: Move invocation of ExceptionListener to after the failoverMutex is released avoiding deadlock possibility

Previously, the ExceptionListener was invoked whilst the failoverMutex was held, between the
two potential state changes (connection state change and session state change).

This commit reorders the statements so that the ExceptionListner is fired after the failoverMutex
is released. It also means that the ExceptionListener is fired *after* both connection/session
have undergone any state changes. The exceptionListener member is also made thread safe.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1514664 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 244 insertions, 0 deletions

diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/connection/ExceptionListenerTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/connection/ExceptionListenerTest.java
new file mode 100644
index 0000000000..141de1e5a8
--- /dev/null
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/connection/ExceptionListenerTest.java
@@ -0,0 +1,244 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one
+ *  or more contributor license agreements.  See the NOTICE file
+ *  distributed with this work for additional information
+ *  regarding copyright ownership.  The ASF licenses this file
+ *  to you under the Apache License, Version 2.0 (the
+ *  "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing,
+ *  software distributed under the License is distributed on an
+ *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ *  KIND, either express or implied.  See the License for the
+ *  specific language governing permissions and limitations
+ *  under the License.
+ *
+ *
+ */
+package org.apache.qpid.test.unit.client.connection;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.jms.Connection;
+import javax.jms.ExceptionListener;
+import javax.jms.IllegalStateException;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.Queue;
+import javax.jms.Session;
+
+import org.apache.qpid.AMQConnectionClosedException;
+import org.apache.qpid.client.AMQNoRouteException;
+import org.apache.qpid.jms.ConnectionURL;
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+import org.apache.qpid.transport.ConnectionException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ExceptionListenerTest extends QpidBrokerTestCase
+{
+    private static final Logger LOGGER = LoggerFactory.getLogger(ExceptionListenerTest.class);
+
+    private volatile Throwable _lastExceptionListenerException = null;
+
+    public void testExceptionListenerHearsBrokerShutdown() throws  Exception
+    {
+        final CountDownLatch exceptionReceivedLatch  = new CountDownLatch(1);
+        final AtomicInteger exceptionCounter = new AtomicInteger(0);
+        final ExceptionListener listener = new ExceptionListener()
+        {
+            public void onException(JMSException exception)
+            {
+                exceptionCounter.incrementAndGet();
+                _lastExceptionListenerException = exception;
+                exceptionReceivedLatch.countDown();
+            }
+        };
+
+        Connection connection = getConnection();
+        connection.setExceptionListener(listener);
+
+        stopBroker();
+
+        exceptionReceivedLatch.await(10, TimeUnit.SECONDS);
+
+        assertEquals("Unexpected number of exceptions received", 1, exceptionCounter.intValue());
+        LOGGER.debug("exception was", _lastExceptionListenerException);
+        assertNotNull("Exception should have cause", _lastExceptionListenerException.getCause());
+        Class<? extends Exception> expectedExceptionClass = isBroker010() ? ConnectionException.class : AMQConnectionClosedException.class;
+        assertEquals(expectedExceptionClass, _lastExceptionListenerException.getCause().getClass());
+    }
+
+    /**
+     * It is reasonable for an application to perform Connection#close within the exception
+     * listener.  This test verifies that close is allowed, and proceeds without generating
+     * further exceptions.
+     */
+    public void testExceptionListenerClosesConnection_IsAllowed() throws  Exception
+    {
+        final CountDownLatch exceptionReceivedLatch  = new CountDownLatch(1);
+        final Connection connection = getConnection();
+        final ExceptionListener listener = new ExceptionListener()
+        {
+            public void onException(JMSException exception)
+            {
+                try
+                {
+                    connection.close();
+                    // PASS
+                }
+                catch (Throwable t)
+                {
+                    _lastExceptionListenerException = t;
+                }
+                finally
+                {
+                    exceptionReceivedLatch.countDown();
+                }
+            }
+        };
+        connection.setExceptionListener(listener);
+
+
+        stopBroker();
+
+        boolean exceptionReceived = exceptionReceivedLatch.await(10, TimeUnit.SECONDS);
+        assertTrue("Exception listener did not hear exception within timeout", exceptionReceived);
+        assertNull("Connection#close() should not have thrown exception", _lastExceptionListenerException);
+    }
+
+    /**
+     * Spring's SingleConnectionFactory installs an ExceptionListener that calls stop()
+     * and ignores any IllegalStateException that result.  This test serves to test this
+     * scenario.
+     */
+    public void testExceptionListenerStopsConnection_ThrowsIllegalStateException() throws  Exception
+    {
+        final CountDownLatch exceptionReceivedLatch  = new CountDownLatch(1);
+        final Connection connection = getConnection();
+        final ExceptionListener listener = new ExceptionListener()
+        {
+            public void onException(JMSException exception)
+            {
+                try
+                {
+                    connection.stop();
+                    fail("Exception not thrown");
+                }
+                catch (IllegalStateException ise)
+                {
+                    // PASS
+                }
+                catch (Throwable t)
+                {
+                    _lastExceptionListenerException = t;
+                }
+                finally
+                {
+                    exceptionReceivedLatch.countDown();
+                }
+            }
+        };
+        connection.setExceptionListener(listener);
+
+        stopBroker();
+
+        boolean exceptionReceived = exceptionReceivedLatch.await(10, TimeUnit.SECONDS);
+        assertTrue("Exception listener did not hear exception within timeout", exceptionReceived);
+        assertNull("Connection#stop() should not have thrown unexpected exception", _lastExceptionListenerException);
+    }
+
+    /**
+     * This test reproduces a deadlock that was the subject of a support call. A Spring based
+     * application was using SingleConnectionFactory.  It installed an ExceptionListener that
+     * stops and closes the connection in response to any exception.  On receipt of a message
+     * the application would create a new session then send a response message (within onMessage).
+     * It appears that a misconfiguration in the application meant that some of these messages
+     * were bounced (no-route). Bounces are treated like connection exceptions and are passed
+     * back to the application via the ExceptionListener.  The deadlock occurred between the
+     * ExceptionListener's call to stop() and the MessageListener's attempt to create a new
+     * session.
+     */
+    public void testExceptionListenerConnectionStopDeadlock() throws  Exception
+    {
+        Queue messageQueue = getTestQueue();
+
+        Map<String, String> options = new HashMap<String, String>();
+        options.put(ConnectionURL.OPTIONS_CLOSE_WHEN_NO_ROUTE, Boolean.toString(false));
+
+        final Connection connection = getConnectionWithOptions(options);
+
+        Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+        session.createConsumer(messageQueue).close(); // Create queue by side-effect
+
+        // Put 10 messages onto messageQueue
+        sendMessage(session, messageQueue, 10);
+
+        // Install an exception listener that stops/closes the connection on receipt of 2nd AMQNoRouteException.
+        // (Triggering on the 2nd (rather than 1st) seems to increase the probability that the test ends in deadlock,
+        // at least on my machine).
+        final CountDownLatch exceptionReceivedLatch  = new CountDownLatch(2);
+        final ExceptionListener listener = new ExceptionListener()
+        {
+            public void onException(JMSException exception)
+            {
+                try
+                {
+                    assertNotNull("JMS Exception must have cause", exception.getCause() );
+                    assertEquals("JMS Exception is of wrong type", AMQNoRouteException.class, exception.getCause().getClass());
+                    exceptionReceivedLatch.countDown();
+                    if (exceptionReceivedLatch.getCount() == 0)
+                    {
+                        connection.stop(); // ** Deadlock
+                        connection.close();
+                    }
+                }
+                catch (Throwable t)
+                {
+                    _lastExceptionListenerException = t;
+                }
+            }
+        };
+        connection.setExceptionListener(listener);
+
+        // Create a message listener that receives from testQueue and tries to forward them to unknown queue (thus
+        // provoking AMQNoRouteException exceptions to be delivered to the ExceptionListener).
+        final Queue unknownQueue = session.createQueue(getTestQueueName() + "_unknown");;
+        MessageListener redirectingMessageListener = new MessageListener()
+        {
+            @Override
+            public void onMessage(Message msg)
+            {
+                try
+                {
+                    Session mlSession = connection.createSession(true, Session.SESSION_TRANSACTED);  // ** Deadlock
+                    mlSession.createProducer(unknownQueue).send(msg);
+                    mlSession.commit();
+                }
+                catch (JMSException je)
+                {
+                    // Connection is closed by the listener, so exceptions here are expected.
+                    LOGGER.debug("Expected exception - message listener got exception", je);
+                }
+            }
+        };
+
+        MessageConsumer consumer = session.createConsumer(messageQueue);
+        consumer.setMessageListener(redirectingMessageListener);
+        connection.start();
+
+        // Await the 2nd exception
+        boolean exceptionReceived = exceptionReceivedLatch.await(10, TimeUnit.SECONDS);
+        assertTrue("Exception listener did not hear exception within timeout", exceptionReceived);
+        assertNull("Exception listener should not have had experienced exception", _lastExceptionListenerException);
+    }
+}