You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ta...@apache.org on 2018/06/06 21:52:20 UTC
qpid-jms git commit: QPIDJMS-386 Handle errors on link recreate
during failover
Repository: qpid-jms
Updated Branches:
refs/heads/master b66120798 -> 91ababa95
QPIDJMS-386 Handle errors on link recreate during failover
Better handle link recreation failures during failover so that on
successive attempts to failover the links are tried again and not left
in an unknown state.
Project: http://git-wip-us.apache.org/repos/asf/qpid-jms/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-jms/commit/91ababa9
Tree: http://git-wip-us.apache.org/repos/asf/qpid-jms/tree/91ababa9
Diff: http://git-wip-us.apache.org/repos/asf/qpid-jms/diff/91ababa9
Branch: refs/heads/master
Commit: 91ababa9500813886c41170e73984b0245e33ac3
Parents: b661207
Author: Timothy Bish <ta...@gmail.com>
Authored: Wed Jun 6 17:50:48 2018 -0400
Committer: Timothy Bish <ta...@gmail.com>
Committed: Wed Jun 6 17:50:48 2018 -0400
----------------------------------------------------------------------
.../java/org/apache/qpid/jms/JmsConnection.java | 12 +-
.../apache/qpid/jms/JmsConnectionFactory.java | 29 ++
.../org/apache/qpid/jms/JmsMessageConsumer.java | 17 +-
.../org/apache/qpid/jms/JmsMessageProducer.java | 15 +-
.../java/org/apache/qpid/jms/JmsSession.java | 2 +-
.../qpid/jms/meta/JmsAbstractResource.java | 4 +-
.../apache/qpid/jms/meta/JmsConnectionInfo.java | 10 +
.../org/apache/qpid/jms/meta/JmsResource.java | 3 +-
.../jms/provider/amqp/AmqpAbstractResource.java | 2 +
.../amqp/builders/AmqpResourceBuilder.java | 2 +-
.../failover/FailoverIntegrationTest.java | 483 ++++++++++++++++++-
11 files changed, 562 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91ababa9/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnection.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnection.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnection.java
index 3b2959e..9f828ab 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnection.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnection.java
@@ -1170,6 +1170,14 @@ public class JmsConnection implements AutoCloseable, Connection, TopicConnection
return connectionInfo.isUseDaemonThread();
}
+ public boolean isCloseLinksThatFailOnReconnect() {
+ return connectionInfo.isCloseLinksThatFailOnReconnect();
+ }
+
+ public void setCloseLinksThatFailOnReconnect(boolean closeLinksThatFailOnReconnect) {
+ connectionInfo.setCloseLinksThatFailOnReconnect(closeLinksThatFailOnReconnect);
+ }
+
//----- Async event handlers ---------------------------------------------//
@Override
@@ -1264,7 +1272,7 @@ public class JmsConnection implements AutoCloseable, Connection, TopicConnection
for (JmsConnectionConsumer connectionConsumer : connectionConsumers.values()) {
JmsConsumerInfo consumerInfo = connectionConsumer.getConsumerInfo();
- if (consumerInfo.isOpen()) {
+ if (!consumerInfo.isClosed()) {
request = new ProviderFuture();
provider.create(consumerInfo, request);
request.sync();
@@ -1285,7 +1293,7 @@ public class JmsConnection implements AutoCloseable, Connection, TopicConnection
for (JmsConnectionConsumer connectionConsumer : connectionConsumers.values()) {
JmsConsumerInfo consumerInfo = connectionConsumer.getConsumerInfo();
- if (consumerInfo.isOpen()) {
+ if (!consumerInfo.isClosed()) {
ProviderFuture request = new ProviderFuture();
provider.start(consumerInfo, request);
request.sync();
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91ababa9/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnectionFactory.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnectionFactory.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnectionFactory.java
index baecfa9..eb0427d 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnectionFactory.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnectionFactory.java
@@ -29,6 +29,7 @@ import javax.jms.ConnectionFactory;
import javax.jms.ExceptionListener;
import javax.jms.JMSContext;
import javax.jms.JMSException;
+import javax.jms.MessageListener;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.TopicConnection;
@@ -87,6 +88,7 @@ public class JmsConnectionFactory extends JNDIStorable implements ConnectionFact
private boolean receiveLocalOnly;
private boolean receiveNoWaitLocalOnly;
private boolean populateJMSXUserID;
+ private boolean closeLinksThatFailOnReconnect;
private String queuePrefix = null;
private String topicPrefix = null;
private boolean validatePropertyNames = true;
@@ -914,6 +916,33 @@ public class JmsConnectionFactory extends JNDIStorable implements ConnectionFact
this.useDaemonThread = useDaemonThread;
}
+
+ /**
+ * @return whether links that fail to be created during failover reconnect are closed or not.
+ */
+ public boolean isCloseLinksThatFailOnReconnect() {
+ return closeLinksThatFailOnReconnect;
+ }
+
+ /**
+ * Controls how the client manages errors on recreation of a link (producer / consumer) during
+ * a failover reconnect attempt (defaults to false).
+ * <p>
+ * When false the failure of a link recreation operation while reestablishing a failed connection
+ * results in the client failing that reconnect attempt and retrying the entire connection process
+ * again. This can be disabled by setting this option to true in which case the client will close
+ * the producer or consumer associated with the failed link create attempt and continue rebuilding
+ * the client resources for the newly reestablished connection. When failing a consumer link the
+ * client will trigger the {@link ExceptionListener} assigned to the Connection if the link that failed
+ * was a consumer and that consumer had an associated JMS {@link MessageListener}.
+ *
+ * @param closeLinksThatFailOnReconnect
+ * whether to close links that fail to establish on failover reconnect.
+ */
+ public void setCloseLinksThatFailOnReconnect(boolean closeLinksThatFailOnReconnect) {
+ this.closeLinksThatFailOnReconnect = closeLinksThatFailOnReconnect;
+ }
+
/**
* Provides an entry point for extensions to be configured on this {@link ConnectionFactory}.
* <p>
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91ababa9/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java
index 7a21e73..e922679 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java
@@ -18,6 +18,7 @@ package org.apache.qpid.jms;
import static org.apache.qpid.jms.message.JmsMessageSupport.lookupAckTypeForDisposition;
+import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
@@ -649,15 +650,23 @@ public class JmsMessageConsumer implements AutoCloseable, MessageConsumer, JmsMe
}
protected void onConnectionRecovery(Provider provider) throws Exception {
- if (consumerInfo.isOpen()) {
+ if (!consumerInfo.isClosed()) {
ProviderFuture request = new ProviderFuture();
- provider.create(consumerInfo, request);
- request.sync();
+ try {
+ provider.create(consumerInfo, request);
+ request.sync();
+ } catch (IOException ioe) {
+ if (connection.isCloseLinksThatFailOnReconnect()) {
+ session.consumerClosed(consumerInfo, ioe);
+ } else {
+ throw ioe;
+ }
+ }
}
}
protected void onConnectionRecovered(Provider provider) throws Exception {
- if (consumerInfo.isOpen()) {
+ if (!consumerInfo.isClosed()) {
ProviderFuture request = new ProviderFuture();
provider.start(consumerInfo, request);
request.sync();
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91ababa9/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageProducer.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageProducer.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageProducer.java
index 1164d15..39f1888 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageProducer.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageProducer.java
@@ -16,6 +16,7 @@
*/
package org.apache.qpid.jms;
+import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
@@ -355,10 +356,18 @@ public class JmsMessageProducer implements AutoCloseable, MessageProducer {
}
protected void onConnectionRecovery(Provider provider) throws Exception {
- if (producerInfo.isOpen()) {
+ if (!producerInfo.isClosed()) {
ProviderFuture request = new ProviderFuture();
- provider.create(producerInfo, request);
- request.sync();
+ try {
+ provider.create(producerInfo, request);
+ request.sync();
+ } catch (IOException ioe) {
+ if (connection.isCloseLinksThatFailOnReconnect()) {
+ session.producerClosed(producerInfo, ioe);
+ } else {
+ throw ioe;
+ }
+ }
}
}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91ababa9/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
index 77df75c..782050e 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
@@ -1337,7 +1337,7 @@ public class JmsSession implements AutoCloseable, Session, QueueSession, TopicSe
}
protected void onConnectionRecovery(Provider provider) throws Exception {
- if (sessionInfo.isOpen()) {
+ if (!sessionInfo.isClosed()) {
ProviderFuture request = new ProviderFuture();
provider.create(sessionInfo, request);
request.sync();
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91ababa9/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsAbstractResource.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsAbstractResource.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsAbstractResource.java
index 548a6cf..a4f04eb 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsAbstractResource.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsAbstractResource.java
@@ -33,7 +33,7 @@ public abstract class JmsAbstractResource implements JmsResource {
this.state = state;
}
- public boolean isOpen() {
- return ResourceState.OPEN.equals(state);
+ public boolean isClosed() {
+ return ResourceState.CLOSED.equals(state);
}
}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91ababa9/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsConnectionInfo.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsConnectionInfo.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsConnectionInfo.java
index f40bef1..13f14ab 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsConnectionInfo.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsConnectionInfo.java
@@ -69,6 +69,7 @@ public final class JmsConnectionInfo extends JmsAbstractResource implements Comp
private boolean populateJMSXUserID;
private boolean useDaemonThread;
private boolean awaitClientID = true;
+ private boolean closeLinksThatFailOnReconnect;
private long sendTimeout = DEFAULT_SEND_TIMEOUT;
private long requestTimeout = DEFAULT_REQUEST_TIMEOUT;
private long connectTimeout = DEFAULT_CONNECT_TIMEOUT;
@@ -114,6 +115,7 @@ public final class JmsConnectionInfo extends JmsAbstractResource implements Comp
copy.connectTimeout = connectTimeout;
copy.validatePropertyNames = validatePropertyNames;
copy.useDaemonThread = useDaemonThread;
+ copy.closeLinksThatFailOnReconnect = closeLinksThatFailOnReconnect;
copy.messageIDPolicy = getMessageIDPolicy().copy();
copy.prefetchPolicy = getPrefetchPolicy().copy();
copy.redeliveryPolicy = getRedeliveryPolicy().copy();
@@ -370,6 +372,14 @@ public final class JmsConnectionInfo extends JmsAbstractResource implements Comp
this.awaitClientID = awaitClientID;
}
+ public boolean isCloseLinksThatFailOnReconnect() {
+ return closeLinksThatFailOnReconnect;
+ }
+
+ public void setCloseLinksThatFailOnReconnect(boolean closeLinksThatFailOnReconnect) {
+ this.closeLinksThatFailOnReconnect = closeLinksThatFailOnReconnect;
+ }
+
public EnumMap<JmsConnectionExtensions, BiFunction<Connection, URI, Object>> getExtensionMap() {
return extensionMap;
}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91ababa9/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsResource.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsResource.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsResource.java
index 7911902..70d0c98 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsResource.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsResource.java
@@ -25,7 +25,8 @@ public interface JmsResource {
enum ResourceState {
INITIALIZED,
OPEN,
- CLOSED;
+ CLOSED,
+ REMOTELY_CLOSED;
}
/**
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91ababa9/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpAbstractResource.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpAbstractResource.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpAbstractResource.java
index 11a14a0..b6e6925 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpAbstractResource.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpAbstractResource.java
@@ -256,6 +256,8 @@ public abstract class AmqpAbstractResource<R extends JmsResource, E extends Endp
@Override
public void processRemoteClose(AmqpProvider provider) throws IOException {
+ getResourceInfo().setState(ResourceState.REMOTELY_CLOSED);
+
if (isAwaitingClose()) {
closeResource(provider, null, true); // Close was expected so ignore any endpoint errors.
} else {
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91ababa9/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/builders/AmqpResourceBuilder.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/builders/AmqpResourceBuilder.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/builders/AmqpResourceBuilder.java
index b244cfe..9bca051 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/builders/AmqpResourceBuilder.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/builders/AmqpResourceBuilder.java
@@ -175,7 +175,7 @@ public abstract class AmqpResourceBuilder<TARGET extends AmqpResource, PARENT ex
// If the resource being built is closed during the creation process
// then this is always an error.
- resourceInfo.setState(ResourceState.CLOSED);
+ resourceInfo.setState(ResourceState.REMOTELY_CLOSED);
// Perform any post processing relating to closure during creation attempt
afterClosed(getResource(), getResourceInfo());
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91ababa9/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java
index 84525b8..7172192 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java
@@ -30,11 +30,11 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-import java.io.IOException;
import java.net.URI;
import java.util.Enumeration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import javax.jms.CompletionListener;
@@ -1934,8 +1934,7 @@ public class FailoverIntegrationTest extends QpidJmsTestCase {
doRemotelyCloseConsumerWithMessageListenerFiresJMSExceptionListenerTestImpl(null, null);
}
- private void doRemotelyCloseConsumerWithMessageListenerFiresJMSExceptionListenerTestImpl(Symbol errorCondition,
- String errorDescription) throws JMSException, InterruptedException, Exception, IOException {
+ private void doRemotelyCloseConsumerWithMessageListenerFiresJMSExceptionListenerTestImpl(Symbol errorCondition, String errorDescription) throws Exception {
try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
CountDownLatch consumerClosed = new CountDownLatch(1);
CountDownLatch exceptionListenerFired = new CountDownLatch(1);
@@ -1943,6 +1942,7 @@ public class FailoverIntegrationTest extends QpidJmsTestCase {
testPeer.expectSaslAnonymous();
testPeer.expectOpen();
testPeer.expectBegin();
+
final JmsConnection connection = establishAnonymousConnecton("failover.maxReconnectAttempts=1", testPeer);
connection.setExceptionListener(new ExceptionListener() {
@Override
@@ -2007,6 +2007,483 @@ public class FailoverIntegrationTest extends QpidJmsTestCase {
}
}
+ @Test(timeout = 20000)
+ public void testFailoverCannotRecreateConsumerFailsConnectionAndRetries() throws Exception {
+ Symbol errorCondition = AmqpError.RESOURCE_DELETED;
+ String errorDescription = "testFailoverCannotRecreateConsumerFailsConnectionAndRetries";
+
+ doTestFailoverCannotRecreateConsumerFailsConnectionAndRetries(errorCondition, errorDescription);
+ }
+
+ @Test(timeout = 20000)
+ public void testFailoverCannotRecreateConsumerFailsConnectionAndRetriesNoErrorConditionGiven() throws Exception {
+ doTestFailoverCannotRecreateConsumerFailsConnectionAndRetries(null, null);
+ }
+
+ private void doTestFailoverCannotRecreateConsumerFailsConnectionAndRetries(Symbol errorCondition, String errorMessage) throws Exception {
+ try (TestAmqpPeer originalPeer = new TestAmqpPeer();
+ TestAmqpPeer rejectingPeer = new TestAmqpPeer();
+ TestAmqpPeer finalPeer = new TestAmqpPeer();) {
+
+ final CountDownLatch originalConnected = new CountDownLatch(1);
+ final CountDownLatch finalConnected = new CountDownLatch(1);
+ final AtomicBoolean exceptionListenerFired = new AtomicBoolean();
+
+ // Create a peer to connect to, then one to reconnect to
+ final String originalURI = createPeerURI(originalPeer);
+ final String rejectingURI = createPeerURI(rejectingPeer);
+ final String finalURI = createPeerURI(finalPeer);
+
+ LOG.info("Original peer is at: {}", originalURI);
+ LOG.info("Rejecting peer is at: {}", rejectingURI);
+ LOG.info("Final peer is at: {}", finalURI);
+
+ // Expect connection to the first peer (and have it drop)
+ originalPeer.expectSaslAnonymous();
+ originalPeer.expectOpen();
+ originalPeer.expectBegin();
+ originalPeer.expectBegin();
+ originalPeer.expectReceiverAttach();
+ originalPeer.expectLinkFlow();
+ originalPeer.dropAfterLastHandler();
+
+ // --- Post Failover Expectations of Rejecting --- //
+ rejectingPeer.expectSaslAnonymous();
+ rejectingPeer.expectOpen();
+ rejectingPeer.expectBegin();
+ rejectingPeer.expectBegin();
+ rejectingPeer.expectReceiverAttach(notNullValue(), notNullValue(), false, true, false, false, errorCondition, errorMessage);
+ // --- Client will clean up connection and then reconnect to next peer --- //
+ rejectingPeer.expectDetach(true, false, false);
+ rejectingPeer.expectClose();
+
+ // --- Post Failover Expectations of FinalPeer --- //
+ finalPeer.expectSaslAnonymous();
+ finalPeer.expectOpen();
+ finalPeer.expectBegin();
+ finalPeer.expectBegin();
+ finalPeer.expectReceiverAttach();
+ finalPeer.expectLinkFlow();
+
+ final JmsConnection connection = establishAnonymousConnecton(originalPeer, rejectingPeer, finalPeer);
+ connection.setExceptionListener(new ExceptionListener() {
+ @Override
+ public void onException(JMSException exception) {
+ LOG.trace("JMS ExceptionListener: ", exception);
+ exceptionListenerFired.set(true);
+ }
+ });
+
+ connection.addConnectionListener(new JmsDefaultConnectionListener() {
+ @Override
+ public void onConnectionEstablished(URI remoteURI) {
+ LOG.info("Connection Established: {}", remoteURI);
+ if (originalURI.equals(remoteURI.toString())) {
+ originalConnected.countDown();
+ }
+ }
+
+ @Override
+ public void onConnectionRestored(URI remoteURI) {
+ LOG.info("Connection Restored: {}", remoteURI);
+ if (finalURI.equals(remoteURI.toString())) {
+ finalConnected.countDown();
+ }
+ }
+ });
+ connection.start();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue("myQueue");
+ final MessageConsumer consumer = session.createConsumer(queue);
+ consumer.setMessageListener(new MessageListener() {
+ @Override
+ public void onMessage(Message message) {
+ }
+ });
+
+ finalPeer.waitForAllHandlersToComplete(1000);
+
+ assertTrue("Should connect to original peer", originalConnected.await(3, TimeUnit.SECONDS));
+ assertTrue("Should connect to final peer", finalConnected.await(3, TimeUnit.SECONDS));
+ assertFalse("The ExceptionListener should not have been alerted", exceptionListenerFired.get());
+
+ // Check that consumer isn't closed
+ try {
+ consumer.getMessageListener();
+ } catch (JMSException ex) {
+ fail("Consumer should be in open state and not throw here.");
+ }
+
+ // Shut it down
+ finalPeer.expectClose();
+ connection.close();
+
+ finalPeer.waitForAllHandlersToComplete(1000);
+ }
+ }
+
+ @Test(timeout = 20000)
+ public void testFailoverCannotRecreateProducerFailsConnectionAndRetries() throws Exception {
+ Symbol errorCondition = AmqpError.RESOURCE_DELETED;
+ String errorDescription = "testFailoverCannotRecreateProducerFailsConnectionAndRetries";
+
+ doTestFailoverCannotRecreateProducerFailsConnectionAndRetries(errorCondition, errorDescription);
+ }
+
+ @Test(timeout = 20000)
+ public void testFailoverCannotRecreateProducerFailsConnectionAndRetriesNoErrorConditionGiven() throws Exception {
+ doTestFailoverCannotRecreateProducerFailsConnectionAndRetries(null, null);
+ }
+
+ private void doTestFailoverCannotRecreateProducerFailsConnectionAndRetries(Symbol errorCondition, String errorMessage) throws Exception {
+ try (TestAmqpPeer originalPeer = new TestAmqpPeer();
+ TestAmqpPeer rejectingPeer = new TestAmqpPeer();
+ TestAmqpPeer finalPeer = new TestAmqpPeer();) {
+
+ final CountDownLatch originalConnected = new CountDownLatch(1);
+ final CountDownLatch finalConnected = new CountDownLatch(1);
+ final AtomicBoolean exceptionListenerFired = new AtomicBoolean();
+
+ // Create a peer to connect to, then one to reconnect to
+ final String originalURI = createPeerURI(originalPeer);
+ final String rejectingURI = createPeerURI(rejectingPeer);
+ final String finalURI = createPeerURI(finalPeer);
+
+ LOG.info("Original peer is at: {}", originalURI);
+ LOG.info("Rejecting peer is at: {}", rejectingURI);
+ LOG.info("Final peer is at: {}", finalURI);
+
+ // Expect connection to the first peer (and have it drop)
+ originalPeer.expectSaslAnonymous();
+ originalPeer.expectOpen();
+ originalPeer.expectBegin();
+ originalPeer.expectBegin();
+ originalPeer.expectSenderAttach();
+ originalPeer.dropAfterLastHandler();
+
+ // --- Post Failover Expectations of Rejecting --- //
+ rejectingPeer.expectSaslAnonymous();
+ rejectingPeer.expectOpen();
+ rejectingPeer.expectBegin();
+ rejectingPeer.expectBegin();
+ rejectingPeer.expectSenderAttach(notNullValue(), notNullValue(), true, false, false, -1, errorCondition, errorMessage);
+ // --- Client will clean up connection and then reconnect to next peer --- //
+ rejectingPeer.expectDetach(true, false, false);
+ rejectingPeer.expectClose();
+
+ // --- Post Failover Expectations of FinalPeer --- //
+ finalPeer.expectSaslAnonymous();
+ finalPeer.expectOpen();
+ finalPeer.expectBegin();
+ finalPeer.expectBegin();
+ finalPeer.expectSenderAttach();
+
+ final JmsConnection connection = establishAnonymousConnecton(originalPeer, rejectingPeer, finalPeer);
+ connection.setExceptionListener(new ExceptionListener() {
+ @Override
+ public void onException(JMSException exception) {
+ LOG.trace("JMS ExceptionListener: ", exception);
+ exceptionListenerFired.set(true);
+ }
+ });
+
+ connection.addConnectionListener(new JmsDefaultConnectionListener() {
+ @Override
+ public void onConnectionEstablished(URI remoteURI) {
+ LOG.info("Connection Established: {}", remoteURI);
+ if (originalURI.equals(remoteURI.toString())) {
+ originalConnected.countDown();
+ }
+ }
+
+ @Override
+ public void onConnectionRestored(URI remoteURI) {
+ LOG.info("Connection Restored: {}", remoteURI);
+ if (finalURI.equals(remoteURI.toString())) {
+ finalConnected.countDown();
+ }
+ }
+ });
+ connection.start();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue("myQueue");
+ final MessageProducer producer = session.createProducer(queue);
+
+ finalPeer.waitForAllHandlersToComplete(1000);
+
+ assertTrue("Should connect to original peer", originalConnected.await(3, TimeUnit.SECONDS));
+ assertTrue("Should connect to final peer", finalConnected.await(3, TimeUnit.SECONDS));
+ assertFalse("The ExceptionListener should not have been alerted", exceptionListenerFired.get());
+
+ // Check that producer isn't closed
+ try {
+ producer.getDestination();
+ } catch (JMSException ex) {
+ fail("Producer should be in open state and not throw here.");
+ }
+
+ // Shut it down
+ finalPeer.expectClose();
+ connection.close();
+
+ finalPeer.waitForAllHandlersToComplete(1000);
+ }
+ }
+
+ @Test(timeout = 20000)
+ public void testFailoverCannotRecreateConsumerWithCloseFailedLinksEnabled() throws Exception {
+ Symbol errorCondition = AmqpError.RESOURCE_DELETED;
+ String errorDescription = "testFailoverCannotRecreateConsumerWithCloseFailedLinksEnabled";
+
+ doTestFailoverCannotRecreateConsumerWithCloseFailedLinksEnabled(true, errorCondition, errorDescription);
+ }
+
+ @Test(timeout = 20000)
+ public void testFailoverCannotRecreateConsumerWithCloseFailedLinksEnabledNoMessageListener() throws Exception {
+ Symbol errorCondition = AmqpError.RESOURCE_DELETED;
+ String errorDescription = "testFailoverCannotRecreateConsumerWithCloseFailedLinksEnabled";
+
+ doTestFailoverCannotRecreateConsumerWithCloseFailedLinksEnabled(false, errorCondition, errorDescription);
+ }
+
+ @Test(timeout = 20000)
+ public void testFailoverCannotRecreateConsumerWithCloseFailedLinksEnabledNoErrorConditionGiven() throws Exception {
+ doTestFailoverCannotRecreateConsumerWithCloseFailedLinksEnabled(true, null, null);
+ }
+
+ @Test(timeout = 20000)
+ public void testFailoverCannotRecreateConsumerWithCloseFailedLinksEnabledNoErrorConditionGivenNoMessageListener() throws Exception {
+ doTestFailoverCannotRecreateConsumerWithCloseFailedLinksEnabled(false, null, null);
+ }
+
+ private void doTestFailoverCannotRecreateConsumerWithCloseFailedLinksEnabled(boolean addListener, Symbol errorCondition, String errorDescription) throws Exception {
+ try (TestAmqpPeer originalPeer = new TestAmqpPeer();
+ TestAmqpPeer finalPeer = new TestAmqpPeer();) {
+
+ final CountDownLatch originalConnected = new CountDownLatch(1);
+ final CountDownLatch finalConnected = new CountDownLatch(1);
+ final CountDownLatch exceptionListenerFired = new CountDownLatch(1);
+
+ // Create a peer to connect to, then one to reconnect to
+ final String originalURI = createPeerURI(originalPeer);
+ final String finalURI = createPeerURI(finalPeer);
+
+ LOG.info("Original peer is at: {}", originalURI);
+ LOG.info("Final peer is at: {}", finalURI);
+
+ // Expect connection to the first peer (and have it drop)
+ originalPeer.expectSaslAnonymous();
+ originalPeer.expectOpen();
+ originalPeer.expectBegin();
+ originalPeer.expectBegin();
+ originalPeer.expectReceiverAttach();
+ originalPeer.expectLinkFlow();
+ originalPeer.dropAfterLastHandler();
+
+ // --- Post Failover Expectations of Rejecting --- //
+ finalPeer.expectSaslAnonymous();
+ finalPeer.expectOpen();
+ finalPeer.expectBegin();
+ finalPeer.expectBegin();
+ finalPeer.expectReceiverAttach(notNullValue(), notNullValue(), false, true, false, false, errorCondition, errorDescription);
+ finalPeer.expectDetach(true, false, false);
+
+ final JmsConnection connection = establishAnonymousConnecton("jms.closeLinksThatFailOnReconnect=true", originalPeer, finalPeer);
+ connection.setExceptionListener(new ExceptionListener() {
+ @Override
+ public void onException(JMSException exception) {
+ LOG.trace("JMS ExceptionListener: ", exception);
+ exceptionListenerFired.countDown();
+ }
+ });
+
+ connection.addConnectionListener(new JmsDefaultConnectionListener() {
+ @Override
+ public void onConnectionEstablished(URI remoteURI) {
+ LOG.info("Connection Established: {}", remoteURI);
+ if (originalURI.equals(remoteURI.toString())) {
+ originalConnected.countDown();
+ }
+ }
+
+ @Override
+ public void onConnectionRestored(URI remoteURI) {
+ LOG.info("Connection Restored: {}", remoteURI);
+ if (finalURI.equals(remoteURI.toString())) {
+ finalConnected.countDown();
+ }
+ }
+ });
+ connection.start();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue("myQueue");
+ final MessageConsumer consumer = session.createConsumer(queue);
+ if (addListener) {
+ consumer.setMessageListener(new MessageListener() {
+ @Override
+ public void onMessage(Message message) {
+ }
+ });
+ }
+
+ finalPeer.waitForAllHandlersToComplete(1000);
+
+ assertTrue("Should connect to original peer", originalConnected.await(3, TimeUnit.SECONDS));
+ assertTrue("Should connect to final peer", finalConnected.await(3, TimeUnit.SECONDS));
+
+ if (addListener) {
+ assertTrue("JMS Exception listener should have fired with a MessageListener", exceptionListenerFired.await(2, TimeUnit.SECONDS));
+
+ // Verify the consumer gets marked closed
+ assertTrue("consumer never closed.", Wait.waitFor(new Wait.Condition() {
+ @Override
+ public boolean isSatisified() throws Exception {
+ try {
+ consumer.getMessageListener();
+ } catch (IllegalStateException jmsise) {
+ if (jmsise.getCause() != null) {
+ String message = jmsise.getCause().getMessage();
+ if (errorCondition != null) {
+ return message.contains(errorCondition.toString()) &&
+ message.contains(errorDescription);
+ } else {
+ return message.contains("Link creation was refused");
+ }
+ } else {
+ return false;
+ }
+ }
+ return false;
+ }
+ }, 5000, 10));
+ } else {
+ assertFalse("The ExceptionListener should not have been alerted", exceptionListenerFired.getCount() == 0);
+ }
+
+ // Shut it down
+ finalPeer.expectClose();
+ connection.close();
+
+ finalPeer.waitForAllHandlersToComplete(1000);
+ }
+ }
+
+ @Test(timeout = 20000)
+ public void testFailoverCannotRecreateProducerWithCloseFailedLinksEnabled() throws Exception {
+ Symbol errorCondition = AmqpError.RESOURCE_DELETED;
+ String errorDescription = "testFailoverCannotRecreateProducerWithCloseFailedLinksEnabled";
+
+ doTestFailoverCannotRecreateWithCloseFailedLinksEnabled(errorCondition, errorDescription);
+ }
+
+ @Test(timeout = 20000)
+ public void testFailoverCannotRecreateProducerWithCloseFailedLinksEnabledNoErrorConditionGiven() throws Exception {
+ doTestFailoverCannotRecreateWithCloseFailedLinksEnabled(null, null);
+ }
+
+ private void doTestFailoverCannotRecreateWithCloseFailedLinksEnabled(Symbol errorCondition, String errorDescription) throws Exception {
+ try (TestAmqpPeer originalPeer = new TestAmqpPeer();
+ TestAmqpPeer finalPeer = new TestAmqpPeer();) {
+
+ final CountDownLatch originalConnected = new CountDownLatch(1);
+ final CountDownLatch finalConnected = new CountDownLatch(1);
+ final AtomicBoolean exceptionListenerFired = new AtomicBoolean();
+
+ // Create a peer to connect to, then one to reconnect to
+ final String originalURI = createPeerURI(originalPeer);
+ final String finalURI = createPeerURI(finalPeer);
+
+ LOG.info("Original peer is at: {}", originalURI);
+ LOG.info("Final peer is at: {}", finalURI);
+
+ // Expect connection to the first peer (and have it drop)
+ originalPeer.expectSaslAnonymous();
+ originalPeer.expectOpen();
+ originalPeer.expectBegin();
+ originalPeer.expectBegin();
+ originalPeer.expectSenderAttach();
+ originalPeer.dropAfterLastHandler();
+
+ // --- Post Failover Expectations of Rejecting --- //
+ finalPeer.expectSaslAnonymous();
+ finalPeer.expectOpen();
+ finalPeer.expectBegin();
+ finalPeer.expectBegin();
+ finalPeer.expectSenderAttach(notNullValue(), notNullValue(), true, false, false, -1, errorCondition, errorDescription);
+ finalPeer.expectDetach(true, false, false);
+
+ final JmsConnection connection = establishAnonymousConnecton("jms.closeLinksThatFailOnReconnect=true", originalPeer, finalPeer);
+ connection.setExceptionListener(new ExceptionListener() {
+ @Override
+ public void onException(JMSException exception) {
+ LOG.trace("JMS ExceptionListener: ", exception);
+ exceptionListenerFired.set(true);
+ }
+ });
+
+ connection.addConnectionListener(new JmsDefaultConnectionListener() {
+ @Override
+ public void onConnectionEstablished(URI remoteURI) {
+ LOG.info("Connection Established: {}", remoteURI);
+ if (originalURI.equals(remoteURI.toString())) {
+ originalConnected.countDown();
+ }
+ }
+
+ @Override
+ public void onConnectionRestored(URI remoteURI) {
+ LOG.info("Connection Restored: {}", remoteURI);
+ if (finalURI.equals(remoteURI.toString())) {
+ finalConnected.countDown();
+ }
+ }
+ });
+ connection.start();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue("myQueue");
+ final MessageProducer producer = session.createProducer(queue);
+
+ finalPeer.waitForAllHandlersToComplete(1000);
+
+ assertTrue("Should connect to original peer", originalConnected.await(3, TimeUnit.SECONDS));
+ assertTrue("Should connect to final peer", finalConnected.await(3, TimeUnit.SECONDS));
+ assertFalse("The ExceptionListener should not have been alerted", exceptionListenerFired.get());
+
+ // Verify the producer gets marked closed
+ assertTrue("producer never closed.", Wait.waitFor(new Wait.Condition() {
+ @Override
+ public boolean isSatisified() throws Exception {
+ try {
+ producer.getDestination();
+ } catch (IllegalStateException jmsise) {
+ if (jmsise.getCause() != null) {
+ String message = jmsise.getCause().getMessage();
+ if (errorCondition != null) {
+ return message.contains(errorCondition.toString()) &&
+ message.contains(errorDescription);
+ } else {
+ return message.contains("Link creation was refused");
+ }
+ } else {
+ return false;
+ }
+ }
+ return false;
+ }
+ }, 5000, 10));
+
+ // Shut it down
+ finalPeer.expectClose();
+ connection.close();
+
+ finalPeer.waitForAllHandlersToComplete(1000);
+ }
+ }
+
private JmsConnection establishAnonymousConnecton(TestAmqpPeer... peers) throws JMSException {
return establishAnonymousConnecton(null, null, peers);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org