From f505cdd862dbd80c2f44331c30f60fbb8432a226 Mon Sep 17 00:00:00 2001 From: Robert Gemmell Date: Fri, 21 Oct 2011 10:14:04 +0000 Subject: QPID-3532: make the 0-10 client hold the failover mutex during the failover. Alter the Address resolution code to allow resolving addresses after failover. Add some more failover tests (inc ADDR based ones). Make the failover process notify any waiters in the session to abort and let failover proceed. Applied patch from Oleksandr Rudyy and myself. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1187279 13f79535-47bb-0310-9956-ffa450edef68 --- .../java/org/apache/qpid/client/AMQConnection.java | 10 + .../qpid/client/AMQConnectionDelegate_0_10.java | 47 ++-- .../org/apache/qpid/client/AMQDestination.java | 18 +- .../org/apache/qpid/client/AMQSession_0_10.java | 6 +- .../qpid/client/BasicMessageProducer_0_10.java | 2 +- .../java/org/apache/qpid/transport/Connection.java | 13 +- .../java/org/apache/qpid/transport/Session.java | 58 ++++- .../AddressBasedFailoverBehaviourTest.java | 14 ++ .../client/failover/FailoverBehaviourTest.java | 276 ++++++++++++++++++++- java/test-profiles/JavaPre010Excludes | 1 + 10 files changed, 404 insertions(+), 41 deletions(-) create mode 100644 java/systests/src/main/java/org/apache/qpid/client/failover/AddressBasedFailoverBehaviourTest.java (limited to 'java') diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java b/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java index f15af72407..941534c7ff 100644 --- a/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java +++ b/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java @@ -175,6 +175,10 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect // new amqp-0-10 encoded format. private boolean _useLegacyMapMessageFormat; + //used to track the last failover time for + //Address resolution purposes + private volatile long _lastFailoverTime = 0; + /** * @param broker brokerdetails * @param username username @@ -1076,6 +1080,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect */ public boolean firePreFailover(boolean redirect) { + _lastFailoverTime = System.currentTimeMillis(); boolean proceed = true; if (_connectionListener != null) { @@ -1462,4 +1467,9 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect } } } + + public long getLastFailoverTime() + { + return _lastFailoverTime; + } } diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java b/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java index 0ed3db6ecb..0d48dd5822 100644 --- a/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java +++ b/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java @@ -281,24 +281,29 @@ public class AMQConnectionDelegate_0_10 implements AMQConnectionDelegate, Connec { _conn.getProtocolHandler().setFailoverLatch(new CountDownLatch(1)); - try + _qpidConnection.notifyFailoverRequired(); + + synchronized (_conn.getFailoverMutex()) { - if (_conn.firePreFailover(false) && _conn.attemptReconnection()) + try { - _conn.failoverPrep(); - _conn.resubscribeSessions(); - _conn.fireFailoverComplete(); - return; + if (_conn.firePreFailover(false) && _conn.attemptReconnection()) + { + _conn.failoverPrep(); + _conn.resubscribeSessions(); + _conn.fireFailoverComplete(); + return; + } + } + catch (Exception e) + { + _logger.error("error during failover", e); + } + finally + { + _conn.getProtocolHandler().getFailoverLatch().countDown(); + _conn.getProtocolHandler().setFailoverLatch(null); } - } - catch (Exception e) - { - _logger.error("error during failover", e); - } - finally - { - _conn.getProtocolHandler().getFailoverLatch().countDown(); - _conn.getProtocolHandler().setFailoverLatch(null); } } @@ -324,6 +329,18 @@ public class AMQConnectionDelegate_0_10 implements AMQConnectionDelegate, Connec public T executeRetrySupport(FailoverProtectedOperation operation) throws E { + if (_conn.isFailingOver()) + { + try + { + _conn.blockUntilNotFailingOver(); + } + catch (InterruptedException e) + { + //ignore + } + } + try { return operation.execute(); diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java b/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java index acd46da11a..f9a38138ba 100644 --- a/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java +++ b/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java @@ -22,6 +22,7 @@ package org.apache.qpid.client; import java.net.URISyntaxException; import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; import javax.jms.Destination; import javax.naming.NamingException; @@ -59,7 +60,7 @@ public abstract class AMQDestination implements Destination, Referenceable private boolean _browseOnly; - private boolean _isAddressResolved; + private AtomicLong _addressResolved = new AtomicLong(0); private AMQShortString _queueName; @@ -77,7 +78,7 @@ public abstract class AMQDestination implements Destination, Referenceable public static final int QUEUE_TYPE = 1; public static final int TOPIC_TYPE = 2; public static final int UNKNOWN_TYPE = 3; - + // ----- Fields required to support new address syntax ------- public enum DestSyntax { @@ -740,12 +741,12 @@ public abstract class AMQDestination implements Destination, Referenceable public boolean isAddressResolved() { - return _isAddressResolved; + return _addressResolved.get() > 0; } - public void setAddressResolved(boolean addressResolved) + public void setAddressResolved(long addressResolved) { - _isAddressResolved = addressResolved; + _addressResolved.set(addressResolved); } private static Address createAddressFromString(String str) @@ -823,7 +824,7 @@ public abstract class AMQDestination implements Destination, Referenceable dest.setTargetNode(_targetNode); dest.setSourceNode(_sourceNode); dest.setLink(_link); - dest.setAddressResolved(_isAddressResolved); + dest.setAddressResolved(_addressResolved.get()); return dest; } @@ -836,4 +837,9 @@ public abstract class AMQDestination implements Destination, Referenceable { _isDurable = b; } + + public boolean isResolvedAfter(long time) + { + return _addressResolved.get() > time; + } } diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java b/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java index c6a64ec894..2869e96a87 100644 --- a/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java +++ b/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java @@ -1179,8 +1179,8 @@ public class AMQSession_0_10 extends AMQSession values = new ArrayList(channels.values()); + for (Session ssn : values) + { + ssn.notifyFailoverRequired(); + } + } } diff --git a/java/common/src/main/java/org/apache/qpid/transport/Session.java b/java/common/src/main/java/org/apache/qpid/transport/Session.java index b732191707..321e5256b2 100644 --- a/java/common/src/main/java/org/apache/qpid/transport/Session.java +++ b/java/common/src/main/java/org/apache/qpid/transport/Session.java @@ -50,6 +50,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; /** * Session @@ -125,6 +126,8 @@ public class Session extends SessionInvoker private SessionDetachCode detachCode; private final Object stateLock = new Object(); + private final AtomicBoolean _failoverRequired = new AtomicBoolean(false); + protected Session(Connection connection, Binary name, long expiry) { this(connection, new SessionDelegate(), name, expiry); @@ -257,6 +260,7 @@ public class Session extends SessionInvoker void resume() { + _failoverRequired.set(false); synchronized (commands) { attach(); @@ -459,7 +463,7 @@ public class Session extends SessionInvoker synchronized (commands) { - if (state == DETACHED || state == CLOSING) + if (state == DETACHED || state == CLOSING || state == CLOSED) { return; } @@ -595,11 +599,12 @@ public class Session extends SessionInvoker if (state != OPEN && state != CLOSED && state != CLOSING) { Thread current = Thread.currentThread(); - if (!current.equals(resumer)) + if (!current.equals(resumer) ) { Waiter w = new Waiter(commands, timeout); while (w.hasTime() && (state != OPEN && state != CLOSED)) { + checkFailoverRequired("Command was interrupted because of failover, before being sent"); w.await(); } } @@ -668,6 +673,7 @@ public class Session extends SessionInvoker } } } + checkFailoverRequired("Command was interrupted because of failover, before being sent"); w.await(); } } @@ -762,6 +768,14 @@ public class Session extends SessionInvoker } } + private void checkFailoverRequired(String message) + { + if (_failoverRequired.get()) + { + throw new SessionException(message); + } + } + protected boolean shouldIssueFlush(int next) { return (next % 65536) == 0; @@ -787,6 +801,7 @@ public class Session extends SessionInvoker Waiter w = new Waiter(commands, timeout); while (w.hasTime() && state != CLOSED && lt(maxComplete, point)) { + checkFailoverRequired("Session sync was interrupted by failover."); log.debug("%s waiting for[%d]: %d, %s", this, point, maxComplete, commands); w.await(); } @@ -847,13 +862,6 @@ public class Session extends SessionInvoker } } - private ConnectionClose close = null; - - void closeCode(ConnectionClose close) - { - this.close = close; - } - ExecutionException getException() { synchronized (results) @@ -904,6 +912,7 @@ public class Session extends SessionInvoker Waiter w = new Waiter(this, timeout); while (w.hasTime() && state != CLOSED && !isDone()) { + checkFailoverRequired("Operation was interrupted by failover."); log.debug("%s waiting for result: %s", Session.this, this); w.await(); } @@ -915,7 +924,12 @@ public class Session extends SessionInvoker } else if (state == CLOSED) { - throw new SessionException(getException()); + ExecutionException ex = getException(); + if(ex == null) + { + throw new SessionClosedException(); + } + throw new SessionException(ex); } else { @@ -995,6 +1009,7 @@ public class Session extends SessionInvoker Waiter w = new Waiter(commands, timeout); while (w.hasTime() && state != CLOSED) { + checkFailoverRequired("close() was interrupted by failover."); w.await(); } @@ -1089,6 +1104,7 @@ public class Session extends SessionInvoker Waiter w = new Waiter(stateLock, timeout); while (w.hasTime() && state == NEW) { + checkFailoverRequired("Session opening was interrupted by failover."); w.await(); } } @@ -1111,4 +1127,26 @@ public class Session extends SessionInvoker { return stateLock; } + + protected void notifyFailoverRequired() + { + //ensure any operations waiting are aborted to + //prevent them waiting for timeout for 60 seconds + //and possibly preventing failover proceeding + _failoverRequired.set(true); + synchronized (commands) + { + commands.notifyAll(); + } + synchronized (results) + { + for (ResultFuture result : results.values()) + { + synchronized(result) + { + result.notifyAll(); + } + } + } + } } diff --git a/java/systests/src/main/java/org/apache/qpid/client/failover/AddressBasedFailoverBehaviourTest.java b/java/systests/src/main/java/org/apache/qpid/client/failover/AddressBasedFailoverBehaviourTest.java new file mode 100644 index 0000000000..980fc7285d --- /dev/null +++ b/java/systests/src/main/java/org/apache/qpid/client/failover/AddressBasedFailoverBehaviourTest.java @@ -0,0 +1,14 @@ +package org.apache.qpid.client.failover; + +import javax.jms.Destination; +import javax.jms.JMSException; +import javax.jms.Session; + +public class AddressBasedFailoverBehaviourTest extends FailoverBehaviourTest +{ + @Override + protected Destination createDestination(Session session) throws JMSException + { + return session.createQueue("ADDR:" +getTestQueueName() + "_" + System.currentTimeMillis() + "; {create: always}"); + } +} diff --git a/java/systests/src/main/java/org/apache/qpid/client/failover/FailoverBehaviourTest.java b/java/systests/src/main/java/org/apache/qpid/client/failover/FailoverBehaviourTest.java index 66f8fe0546..a5b9c618bc 100644 --- a/java/systests/src/main/java/org/apache/qpid/client/failover/FailoverBehaviourTest.java +++ b/java/systests/src/main/java/org/apache/qpid/client/failover/FailoverBehaviourTest.java @@ -20,6 +20,7 @@ package org.apache.qpid.client.failover; import java.text.MessageFormat; import java.util.ArrayList; +import java.util.Enumeration; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -33,12 +34,15 @@ import javax.jms.Message; import javax.jms.MessageConsumer; import javax.jms.MessageListener; import javax.jms.MessageProducer; +import javax.jms.Queue; +import javax.jms.QueueBrowser; import javax.jms.Session; import javax.jms.TextMessage; import javax.jms.TransactionRolledBackException; import org.apache.qpid.client.AMQConnection; import org.apache.qpid.jms.ConnectionListener; +import org.apache.qpid.jms.FailoverPolicy; import org.apache.qpid.test.utils.FailoverBaseCase; /** @@ -96,6 +100,11 @@ public class FailoverBehaviourTest extends FailoverBaseCase implements Connectio */ private JMSException _exceptionListenerException; + /** + * Latch to check that failover mutex is hold by a failover thread + */ + private CountDownLatch _failoverStarted; + @Override protected void setUp() throws Exception { @@ -105,6 +114,7 @@ public class FailoverBehaviourTest extends FailoverBaseCase implements Connectio _connection.setExceptionListener(this); ((AMQConnection) _connection).setConnectionListener(this); _failoverComplete = new CountDownLatch(1); + _failoverStarted = new CountDownLatch(1); } /** @@ -625,8 +635,134 @@ public class FailoverBehaviourTest extends FailoverBaseCase implements Connectio sessionCloseAfterFailoverImpl(Session.AUTO_ACKNOWLEDGE); } + public void testPublishAutoAcknowledgedWhileFailover() throws Exception + { + publishWhileFailingOver(Session.AUTO_ACKNOWLEDGE); + } + + public void testPublishClientAcknowledgedWhileFailover() throws Exception + { + Message receivedMessage = publishWhileFailingOver(Session.CLIENT_ACKNOWLEDGE); + receivedMessage.acknowledge(); + } + + public void testPublishTransactedAcknowledgedWhileFailover() throws Exception + { + publishWhileFailingOver(Session.SESSION_TRANSACTED); + _consumerSession.commit(); + } + + public void testPublishAutoAcknowledgedWithFailoverMutex() throws Exception + { + publishWithFailoverMutex(Session.AUTO_ACKNOWLEDGE); + } + + public void testPublishClientAcknowledgedWithFailoverMutex() throws Exception + { + publishWithFailoverMutex(Session.CLIENT_ACKNOWLEDGE); + + } + + public void testPublishTransactedAcknowledgedWithFailoverMutex() throws Exception + { + publishWithFailoverMutex(Session.SESSION_TRANSACTED); + } + + public void testClientAcknowledgedSessionCloseWhileFailover() throws Exception + { + sessionCloseWhileFailoverImpl(Session.CLIENT_ACKNOWLEDGE); + } + + public void testTransactedSessionCloseWhileFailover() throws Exception + { + sessionCloseWhileFailoverImpl(Session.SESSION_TRANSACTED); + } + + public void testAutoAcknowledgedSessionCloseWhileFailover() throws Exception + { + sessionCloseWhileFailoverImpl(Session.AUTO_ACKNOWLEDGE); + } + + public void testClientAcknowledgedQueueBrowserCloseWhileFailover() throws Exception + { + browserCloseWhileFailoverImpl(Session.CLIENT_ACKNOWLEDGE); + } + + public void testTransactedQueueBrowserCloseWhileFailover() throws Exception + { + browserCloseWhileFailoverImpl(Session.SESSION_TRANSACTED); + } + + public void testAutoAcknowledgedQueueBrowserCloseWhileFailover() throws Exception + { + browserCloseWhileFailoverImpl(Session.AUTO_ACKNOWLEDGE); + } + + private Message publishWhileFailingOver(int autoAcknowledge) throws JMSException, InterruptedException + { + setDelayedFailoverPolicy(5); + init(autoAcknowledge, true); + + String text = MessageFormat.format(TEST_MESSAGE_FORMAT, 0); + Message message = _producerSession.createTextMessage(text); + + failBroker(getFailingPort()); + + if(!_failoverStarted.await(5, TimeUnit.SECONDS)) + { + fail("Did not receieve notification failover had started"); + } + + _producer.send(message); + + if (_producerSession.getTransacted()) + { + _producerSession.commit(); + } + + Message receivedMessage = _consumer.receive(1000l); + assertReceivedMessage(receivedMessage, TEST_MESSAGE_FORMAT, 0); + return receivedMessage; + } + + private void publishWithFailoverMutex(int autoAcknowledge) throws JMSException, InterruptedException + { + setDelayedFailoverPolicy(5); + init(autoAcknowledge, true); + + String text = MessageFormat.format(TEST_MESSAGE_FORMAT, 0); + Message message = _producerSession.createTextMessage(text); + + AMQConnection connection = (AMQConnection)_connection; + + // holding failover mutex should prevent the failover from + // proceeding before we try to send the message + synchronized(connection.getFailoverMutex()) + { + failBroker(getFailingPort()); + + // wait to make sure that connection is lost + while(!connection.isFailingOver()) + { + Thread.sleep(25l); + } + + try + { + _producer.send(message); + fail("Sending should fail because connection was lost and failover has not yet completed"); + } + catch(JMSException e) + { + // JMSException is expected + } + } + // wait for failover completion, thus ensuring it actually + //got started, before allowing the test to tear down + awaitForFailoverCompletion(DEFAULT_FAILOVER_TIME); + } /** - * Tests {@link Session#close()} for session with given acknowledge mode + * Tests {@link Session#close()} for session with given acknowledge mode * to ensure that close works after failover. * * @param acknowledgeMode session acknowledge mode @@ -671,7 +807,7 @@ public class FailoverBehaviourTest extends FailoverBaseCase implements Connectio boolean isTransacted = acknowledgeMode == Session.SESSION_TRANSACTED ? true : false; _consumerSession = _connection.createSession(isTransacted, acknowledgeMode); - _destination = _consumerSession.createQueue(getTestQueueName() + "_" + System.currentTimeMillis()); + _destination = createDestination(_consumerSession); _consumer = _consumerSession.createConsumer(_destination); if (startConnection) @@ -684,6 +820,11 @@ public class FailoverBehaviourTest extends FailoverBaseCase implements Connectio } + protected Destination createDestination(Session session) throws JMSException + { + return session.createQueue(getTestQueueName() + "_" + System.currentTimeMillis()); + } + /** * Resends messages if reconnected to a non-clustered broker * @@ -879,6 +1020,7 @@ public class FailoverBehaviourTest extends FailoverBaseCase implements Connectio @Override public boolean preFailover(boolean redirect) { + _failoverStarted.countDown(); return true; } @@ -900,6 +1042,39 @@ public class FailoverBehaviourTest extends FailoverBaseCase implements Connectio _exceptionListenerException = e; } + /** + * Causes 1 second delay before reconnect in order to test whether JMS + * methods block while failover is in progress + */ + private static class DelayingFailoverPolicy extends FailoverPolicy + { + + private CountDownLatch _suspendLatch; + private long _delay; + + public DelayingFailoverPolicy(AMQConnection connection, long delay) + { + super(connection.getConnectionURL(), connection); + _suspendLatch = new CountDownLatch(1); + _delay = delay; + } + + public void attainedConnection() + { + try + { + _suspendLatch.await(_delay, TimeUnit.SECONDS); + } + catch (InterruptedException e) + { + // continue + } + super.attainedConnection(); + } + + } + + private class FailoverTestMessageListener implements MessageListener { // message counter @@ -946,4 +1121,101 @@ public class FailoverBehaviourTest extends FailoverBaseCase implements Connectio return _counter.get(); } } + + /** + * Tests {@link Session#close()} for session with given acknowledge mode + * to ensure that it blocks until failover implementation restores connection. + * + * @param acknowledgeMode session acknowledge mode + * @throws JMSException + */ + private void sessionCloseWhileFailoverImpl(int acknowledgeMode) throws JMSException + { + initDelayedFailover(acknowledgeMode); + + // intentionally receive message but not commit or acknowledge it in + // case of transacted or CLIENT_ACK session + Message receivedMessage = _consumer.receive(1000l); + assertReceivedMessage(receivedMessage, TEST_MESSAGE_FORMAT, 0); + + failBroker(getFailingPort()); + + // test whether session#close blocks while failover is in progress + _consumerSession.close(); + + assertFailoverException(); + } + + /** + * A helper method to instantiate {@link QueueBrowser} and publish test messages on a test queue for further browsing. + * + * @param acknowledgeMode session acknowledge mode + * @return queue browser + * @throws JMSException + */ + private QueueBrowser prepareQueueBrowser(int acknowledgeMode) throws JMSException + { + init(acknowledgeMode, false); + _consumer.close(); + QueueBrowser browser = _consumerSession.createBrowser((Queue) _destination); + _connection.start(); + + produceMessages(TEST_MESSAGE_FORMAT, _messageNumber, false); + if (acknowledgeMode == Session.SESSION_TRANSACTED) + { + _producerSession.commit(); + } + return browser; + } + + /** + * Tests {@link QueueBrowser#close()} for session with given acknowledge mode + * to ensure that it blocks until failover implementation restores connection. + * + * @param acknowledgeMode session acknowledge mode + * @throws JMSException + */ + private void browserCloseWhileFailoverImpl(int acknowledgeMode) throws JMSException + { + setDelayedFailoverPolicy(); + + QueueBrowser browser = prepareQueueBrowser(acknowledgeMode); + + @SuppressWarnings("unchecked") + Enumeration messages = browser.getEnumeration(); + Message receivedMessage = (Message) messages.nextElement(); + assertReceivedMessage(receivedMessage, TEST_MESSAGE_FORMAT, 0); + + failBroker(getFailingPort()); + + browser.close(); + + assertFailoverException(); + } + + private DelayingFailoverPolicy initDelayedFailover(int acknowledgeMode) throws JMSException + { + DelayingFailoverPolicy failoverPolicy = setDelayedFailoverPolicy(); + init(acknowledgeMode, true); + produceMessages(TEST_MESSAGE_FORMAT, _messageNumber, false); + if (acknowledgeMode == Session.SESSION_TRANSACTED) + { + _producerSession.commit(); + } + return failoverPolicy; + } + + private DelayingFailoverPolicy setDelayedFailoverPolicy() + { + return setDelayedFailoverPolicy(2); + } + + private DelayingFailoverPolicy setDelayedFailoverPolicy(long delay) + { + AMQConnection amqConnection = (AMQConnection) _connection; + DelayingFailoverPolicy failoverPolicy = new DelayingFailoverPolicy(amqConnection, delay); + ((AMQConnection) _connection).setFailoverPolicy(failoverPolicy); + return failoverPolicy; + } + } diff --git a/java/test-profiles/JavaPre010Excludes b/java/test-profiles/JavaPre010Excludes index 8344cd98c2..68feaf1e2b 100644 --- a/java/test-profiles/JavaPre010Excludes +++ b/java/test-profiles/JavaPre010Excludes @@ -28,6 +28,7 @@ org.apache.qpid.test.client.message.JMSDestinationTest#testReceiveResend org.apache.qpid.test.client.message.JMSDestinationTest#testGetDestinationWithCustomExchange // The new addressing based syntax is not supported for AMQP 0-8/0-9 versions +org.apache.qpid.client.failover.AddressBasedFailoverBehaviourTest#* org.apache.qpid.test.client.destination.AddressBasedDestinationTest#* org.apache.qpid.test.client.queue.QueuePolicyTest#testRingPolicy org.apache.qpid.test.client.queue.QueuePolicyTest#testRejectPolicy -- cgit v1.2.1