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 2024/09/20 15:19:45 UTC
(qpid-jms) branch 1.x updated: QPIDJMS-600 Ensure session and connection close await async sends
This is an automated email from the ASF dual-hosted git repository.
tabish pushed a commit to branch 1.x
in repository https://gitbox.apache.org/repos/asf/qpid-jms.git
The following commit(s) were added to refs/heads/1.x by this push:
new 408fec7f QPIDJMS-600 Ensure session and connection close await async sends
408fec7f is described below
commit 408fec7f3de80d6106e2dbe2ff3f688c8df363e6
Author: Timothy Bish <ta...@gmail.com>
AuthorDate: Fri Apr 19 18:58:03 2024 -0400
QPIDJMS-600 Ensure session and connection close await async sends
Session and Connection close should be awaiting the outcome of async send
completions before returning. This change allows them to await up to the
close timeout value before moving on and failing any completions that are
not completed after that point. Several tests added to cover this behavior.
(cherry picked from commit 90eb60f59cb59b7b9ad8363ee8a843d6903b8e77)
---
.../java/org/apache/qpid/jms/JmsConnection.java | 8 +
.../main/java/org/apache/qpid/jms/JmsSession.java | 65 ++++++--
.../integration/BytesMessageIntegrationTest.java | 2 +-
.../jms/integration/MapMessageIntegrationTest.java | 2 +-
.../jms/integration/MessageIntegrationTest.java | 2 +-
.../integration/ObjectMessageIntegrationTest.java | 1 +
.../jms/integration/ProducerIntegrationTest.java | 163 ++++++++++++++++++++-
.../jms/integration/SessionIntegrationTest.java | 80 ++++++++++
.../integration/StreamMessageIntegrationTest.java | 1 +
.../integration/TextMessageIntegrationTest.java | 1 +
.../provider/failover/FailoverIntegrationTest.java | 63 ++++++++
11 files changed, 365 insertions(+), 23 deletions(-)
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 4fcdefdb..cb076a36 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
@@ -916,6 +916,14 @@ public class JmsConnection implements AutoCloseable, Connection, TopicConnection
}
}
+ ProviderFuture newProviderFuture() {
+ return newProviderFuture(null);
+ }
+
+ ProviderFuture newProviderFuture(ProviderSynchronization synchronization) {
+ return provider.newProviderFuture(synchronization);
+ }
+
//----- Property setters and getters -------------------------------------//
@Override
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 4907d169..d35e0419 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
@@ -126,6 +126,7 @@ public class JmsSession implements AutoCloseable, Session, QueueSession, TopicSe
private final ReentrantLock sendLock = new ReentrantLock();
private volatile ThreadPoolExecutor deliveryExecutor;
private volatile ThreadPoolExecutor completionExcecutor;
+ private volatile ProviderFuture asyncSendsCompletion;
private AtomicReference<Thread> deliveryThread = new AtomicReference<Thread>();
private boolean deliveryThreadCheckEnabled = true;
private AtomicReference<Thread> completionThread = new AtomicReference<Thread>();
@@ -351,6 +352,7 @@ public class JmsSession implements AutoCloseable, Session, QueueSession, TopicSe
for (JmsMessageProducer producer : new ArrayList<JmsMessageProducer>(this.producers.values())) {
producer.shutdown(cause);
}
+
} catch (JMSException jmsEx) {
shutdownError = jmsEx;
}
@@ -367,30 +369,52 @@ public class JmsSession implements AutoCloseable, Session, QueueSession, TopicSe
}
}
- // Ensure that no asynchronous completion sends remain blocked after close.
+ try {
+ if (getSessionMode() == Session.CLIENT_ACKNOWLEDGE) {
+ acknowledge(ACK_TYPE.SESSION_SHUTDOWN);
+ }
+ } catch (Exception e) {
+ LOG.trace("Exception during session shutdown cleanup acknowledgement", e);
+ }
+
+ // Ensure that no asynchronous completion sends remain blocked after close but wait
+ // using the close timeout for the asynchronous sends to complete normally.
+ final ExecutorService completionExecutor = getCompletionExecutor();
+
synchronized (sessionInfo) {
+ // Producers are now quiesced and we can await completion of asynchronous sends
+ // that are still pending a result or timeout once we've done a quick check to
+ // see if any are actually pending or have completed already.
+ asyncSendsCompletion = connection.newProviderFuture();
+
+ completionExecutor.execute(() -> {
+ if (asyncSendQueue.isEmpty()) {
+ asyncSendsCompletion.onSuccess();
+ }
+ });
+ }
+
+ try {
+ asyncSendsCompletion.sync(connection.getCloseTimeout(), TimeUnit.MILLISECONDS);
+ } catch (Exception ex) {
+ LOG.trace("Exception during wait for asynchronous sends to complete", ex);
+ } finally {
if (cause == null) {
cause = new JMSException("Session closed remotely before message transfer result was notified");
}
- getCompletionExecutor().execute(new FailOrCompleteAsyncCompletionsTask(JmsExceptionSupport.create(cause)));
- getCompletionExecutor().shutdown();
+ // as a last task we want to fail any stragglers in the asynchronous send queue and then
+ // shutdown the queue to prevent any more submissions while the cleanup goes on.
+ completionExecutor.execute(new FailOrCompleteAsyncCompletionsTask(JmsExceptionSupport.create(cause)));
+ completionExecutor.shutdown();
}
try {
- getCompletionExecutor().awaitTermination(connection.getCloseTimeout(), TimeUnit.MILLISECONDS);
+ completionExecutor.awaitTermination(connection.getCloseTimeout(), TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
LOG.trace("Session close awaiting send completions was interrupted");
}
- try {
- if (getSessionMode() == Session.CLIENT_ACKNOWLEDGE) {
- acknowledge(ACK_TYPE.SESSION_SHUTDOWN);
- }
- } catch (Exception e) {
- LOG.trace("Exception during session shutdown cleanup acknowledgement", e);
- }
-
if (shutdownError != null) {
throw shutdownError;
}
@@ -856,11 +880,12 @@ public class JmsSession implements AutoCloseable, Session, QueueSession, TopicSe
}
private void send(JmsMessageProducer producer, JmsDestination destination, Message original, int deliveryMode, int priority, long timeToLive, boolean disableMsgId, boolean disableTimestamp, long deliveryDelay, CompletionListener listener) throws JMSException {
- sendLock.lock();
-
JmsMessage outbound = null;
+ sendLock.lock();
try {
+ checkClosed();
+
original.setJMSDeliveryMode(deliveryMode);
original.setJMSPriority(priority);
original.setJMSRedelivered(false);
@@ -909,7 +934,7 @@ public class JmsSession implements AutoCloseable, Session, QueueSession, TopicSe
}
outbound.getFacade().setDeliveryTime(deliveryTime, hasDelay);
- if(!isJmsMessage) {
+ if (!isJmsMessage) {
// If the original was a foreign message, we still need to update it too.
setForeignMessageDeliveryTime(original, deliveryTime);
}
@@ -977,7 +1002,7 @@ public class JmsSession implements AutoCloseable, Session, QueueSession, TopicSe
}
} catch (JMSException jmsEx) {
// Ensure that on failure case the message is returned to usable state for another send attempt.
- if(outbound != null) {
+ if (outbound != null) {
outbound.onSendComplete();
}
throw jmsEx;
@@ -1511,6 +1536,10 @@ public class JmsSession implements AutoCloseable, Session, QueueSession, TopicSe
if (producerId == null) {
asyncSendQueue.clear();
}
+
+ if (closed.get() && asyncSendsCompletion != null && asyncSendQueue.isEmpty()) {
+ asyncSendsCompletion.onSuccess();
+ }
}
}
@@ -1577,6 +1606,10 @@ public class JmsSession implements AutoCloseable, Session, QueueSession, TopicSe
}
}
}
+
+ if (closed.get() && asyncSendsCompletion != null && asyncSendQueue.isEmpty()) {
+ asyncSendsCompletion.onSuccess();
+ }
} catch (Exception ex) {
LOG.error("Async completion task encountered unexpected failure", ex);
}
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/BytesMessageIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/BytesMessageIntegrationTest.java
index 639c5d8f..298d97e3 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/BytesMessageIntegrationTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/BytesMessageIntegrationTest.java
@@ -581,7 +581,7 @@ public class BytesMessageIntegrationTest extends QpidJmsTestCase {
@Timeout(20)
public void testAsyncCompletionSendMarksBytesMessageReadOnly() throws Exception {
try(TestAmqpPeer testPeer = new TestAmqpPeer();) {
- JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer);
+ JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, "jms.closeTimeout=50");
connection.setSendTimeout(15000);
testPeer.expectBegin();
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MapMessageIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MapMessageIntegrationTest.java
index 954938f0..60b9c81c 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MapMessageIntegrationTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MapMessageIntegrationTest.java
@@ -440,7 +440,7 @@ public class MapMessageIntegrationTest extends QpidJmsTestCase {
@Timeout(20)
public void testAsyncCompletionSendMarksMapMessageReadOnly() throws Exception {
try(TestAmqpPeer testPeer = new TestAmqpPeer();) {
- JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer);
+ JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, "jms.closeTimeout=50");
connection.setSendTimeout(15000);
testPeer.expectBegin();
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MessageIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MessageIntegrationTest.java
index 60f3020d..19bcf698 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MessageIntegrationTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/MessageIntegrationTest.java
@@ -2231,7 +2231,7 @@ public class MessageIntegrationTest extends QpidJmsTestCase
@Timeout(20)
public void testAsyncCompletionSendMarksMessageReadOnly() throws Exception {
try(TestAmqpPeer testPeer = new TestAmqpPeer();) {
- JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer);
+ JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, "jms.closeTimeout=50");
connection.setSendTimeout(15000);
testPeer.expectBegin();
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ObjectMessageIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ObjectMessageIntegrationTest.java
index ebe2dcec..5b6911ef 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ObjectMessageIntegrationTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ObjectMessageIntegrationTest.java
@@ -660,6 +660,7 @@ public class ObjectMessageIntegrationTest extends QpidJmsTestCase {
try(TestAmqpPeer testPeer = new TestAmqpPeer();) {
JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer);
connection.setSendTimeout(15000);
+ connection.setCloseTimeout(10);
testPeer.expectBegin();
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java
index e01c6a4f..4dd6b01d 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java
@@ -2322,7 +2322,7 @@ public class ProducerIntegrationTest extends QpidJmsTestCase {
@Timeout(20)
public void testAsyncCompletionGetsNotifiedWhenSessionClosed() throws Exception {
try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
- JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer);
+ JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, "jms.closeTimeout=100");
testPeer.expectBegin();
testPeer.expectSenderAttach();
@@ -2343,6 +2343,8 @@ public class ProducerIntegrationTest extends QpidJmsTestCase {
producer.send(message, listener);
+ assertFalse(listener.hasCompleted()); // Close should complete it as failed on timeout
+
session.close();
assertTrue(listener.awaitCompletion(5, TimeUnit.SECONDS), "Did not get async callback");
@@ -2356,11 +2358,51 @@ public class ProducerIntegrationTest extends QpidJmsTestCase {
}
}
+ @Test
+ @Timeout(20)
+ public void testAsyncCompletionGetsNotifiedWhenSessionClosedAndWaitForCompletion() throws Exception {
+ try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+ JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, "jms.closeTimeout=1000");
+
+ testPeer.expectBegin();
+ testPeer.expectSenderAttach();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue("myQueue");
+
+ MessageProducer producer = session.createProducer(queue);
+
+ // Create and transfer a new message
+ String text = "myMessage";
+ testPeer.expectTransfer(new TransferPayloadCompositeMatcher(), nullValue(), false, true, new Accepted(), true, 0, 100);
+ testPeer.expectEnd();
+ testPeer.expectClose();
+
+ TextMessage message = session.createTextMessage(text);
+ TestJmsCompletionListener listener = new TestJmsCompletionListener();
+
+ producer.send(message, listener);
+
+ assertFalse(listener.hasCompleted()); // Close should complete it as accepted after the delay
+
+ session.close();
+
+ assertTrue(listener.awaitCompletion(5, TimeUnit.SECONDS), "Did not get async callback");
+ assertNull(listener.exception);
+ assertNotNull(listener.message);
+ assertTrue(listener.message instanceof TextMessage);
+
+ connection.close();
+
+ testPeer.waitForAllHandlersToComplete(2000);
+ }
+ }
+
@Test
@Timeout(20)
public void testAsyncCompletionGetsNotifiedWhenConnectionClosed() throws Exception {
try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
- JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer);
+ JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, "jms.closeTimeout=150");
testPeer.expectBegin();
testPeer.expectSenderAttach();
@@ -2380,6 +2422,8 @@ public class ProducerIntegrationTest extends QpidJmsTestCase {
producer.send(message, listener);
+ assertFalse(listener.hasCompleted());
+
connection.close();
assertTrue(listener.awaitCompletion(5, TimeUnit.SECONDS), "Did not get async callback");
@@ -2391,6 +2435,43 @@ public class ProducerIntegrationTest extends QpidJmsTestCase {
}
}
+ @Test
+ @Timeout(20)
+ public void testAsyncCompletionAllowedToCompleteNormallyWhenConnectionClosed() throws Exception {
+ try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+ JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, "jms.closeTimeout=1000");
+
+ testPeer.expectBegin();
+ testPeer.expectSenderAttach();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue("myQueue");
+
+ MessageProducer producer = session.createProducer(queue);
+
+ // Create and transfer a new message
+ String text = "myMessage";
+ testPeer.expectTransfer(new TransferPayloadCompositeMatcher(), nullValue(), false, true, new Accepted(), true, 0, 100);
+ testPeer.expectClose();
+
+ TextMessage message = session.createTextMessage(text);
+ TestJmsCompletionListener listener = new TestJmsCompletionListener();
+
+ producer.send(message, listener);
+
+ assertFalse(listener.hasCompleted());
+
+ connection.close();
+
+ assertTrue(listener.awaitCompletion(5, TimeUnit.SECONDS), "Did not get async callback");
+ assertNull(listener.exception);
+ assertNotNull(listener.message);
+ assertTrue(listener.message instanceof TextMessage);
+
+ testPeer.waitForAllHandlersToComplete(2000);
+ }
+ }
+
@Test
@Timeout(20)
public void testAsyncCompletionResetsBytesMessage() throws Exception {
@@ -2856,7 +2937,7 @@ public class ProducerIntegrationTest extends QpidJmsTestCase {
@Timeout(20)
public void testRemotelyCloseOneProducerDoesNotCompleteAsyncSendFromAnotherProducer() throws Exception {
try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
- JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer);
+ JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, "jms.closeTimeout=150");
testPeer.expectBegin();
testPeer.expectSenderAttach();
@@ -2895,7 +2976,7 @@ public class ProducerIntegrationTest extends QpidJmsTestCase {
assertFalse(listener.awaitCompletion(10, TimeUnit.MILLISECONDS), "Should not get async callback");
- // Closing the session should complete the send with an exception
+ // Closing the session should complete the send with an exception after timeout
testPeer.expectEnd();
session.close();
@@ -3044,6 +3125,10 @@ public class ProducerIntegrationTest extends QpidJmsTestCase {
this.completed = new CountDownLatch(expected);
}
+ public boolean hasCompleted() {
+ return completed.getCount() == 0;
+ }
+
public boolean awaitCompletion(long timeout, TimeUnit units) throws InterruptedException {
return completed.await(timeout, units);
}
@@ -3239,4 +3324,74 @@ public class ProducerIntegrationTest extends QpidJmsTestCase {
testPeer.waitForAllHandlersToComplete(1000);
}
}
+
+ @Test
+ @Timeout(20)
+ public void testRemotelyEndConnectionCompletesAsyncSends() throws Exception {
+ final String BREAD_CRUMB = "ErrorMessage";
+
+ try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+ final CountDownLatch connectionClosed = new CountDownLatch(1);
+ JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer);
+ connection.addConnectionListener(new JmsDefaultConnectionListener() {
+ @Override
+ public void onConnectionFailure(Throwable exception) {
+ connectionClosed.countDown();
+ }
+ });
+
+ testPeer.expectBegin();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ // Create a producer, then remotely end the session afterwards.
+ testPeer.expectSenderAttach();
+
+ Queue queue = session.createQueue("myQueue");
+ final MessageProducer producer = session.createProducer(queue);
+
+ final int MSG_COUNT = 3;
+
+ for (int i = 0; i < MSG_COUNT; ++i) {
+ testPeer.expectTransferButDoNotRespond(new TransferPayloadCompositeMatcher());
+ }
+
+ TestJmsCompletionListener listener = new TestJmsCompletionListener(MSG_COUNT);
+ try {
+ for (int i = 0; i < MSG_COUNT; ++i) {
+ Message message = session.createTextMessage("content");
+ producer.send(message, listener);
+ }
+ } catch (JMSException e) {
+ LOG.warn("Caught unexpected error: {}", e.getMessage());
+ fail("No expected exception for this send.");
+ }
+
+ testPeer.waitForAllHandlersToComplete(2000);
+ testPeer.expectSenderAttach();
+ testPeer.remotelyCloseConnection(true, AmqpError.RESOURCE_DELETED, BREAD_CRUMB, 50);
+
+ session.createProducer(queue);
+
+ // Verify the session gets marked closed
+ assertTrue(connectionClosed.await(5, TimeUnit.SECONDS), "Session closed callback didn't trigger");
+
+ try {
+ producer.getDeliveryMode();
+ fail("Expected ISE to be thrown due to being closed");
+ } catch (IllegalStateException jmsise) {
+ String errorMessage = jmsise.getCause().getMessage();
+ assertTrue(errorMessage.contains(AmqpError.RESOURCE_DELETED.toString()));
+ assertTrue(errorMessage.contains(BREAD_CRUMB));
+ }
+
+ assertTrue(listener.awaitCompletion(5, TimeUnit.SECONDS));
+ assertEquals(MSG_COUNT, listener.errorCount); // All sends should have been failed
+
+ // Try closing it explicitly, should effectively no-op in client.
+ // The test peer will throw during close if it sends anything.
+ connection.close();
+
+ testPeer.waitForAllHandlersToComplete(1000);
+ }
+ }
}
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java
index 98a12eca..c8d98145 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java
@@ -2333,6 +2333,10 @@ public class SessionIntegrationTest extends QpidJmsTestCase {
completed = new CountDownLatch(expected);
}
+ public boolean hasCompleted() {
+ return completed.getCount() == 0;
+ }
+
public boolean awaitCompletion(long timeout, TimeUnit units) throws InterruptedException {
return completed.await(timeout, units);
}
@@ -2836,6 +2840,82 @@ public class SessionIntegrationTest extends QpidJmsTestCase {
}
}
+ @Test
+ @Timeout(20)
+ public void testRemotelyEndSessionCompletesAsyncSends() throws Exception {
+ final String BREAD_CRUMB = "ErrorMessage";
+
+ try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+ final CountDownLatch sessionClosed = new CountDownLatch(1);
+ JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer);
+ connection.addConnectionListener(new JmsDefaultConnectionListener() {
+ @Override
+ public void onSessionClosed(Session session, Throwable exception) {
+ sessionClosed.countDown();
+ }
+ });
+
+ testPeer.expectBegin();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ // Create a producer, then remotely end the session afterwards.
+ testPeer.expectSenderAttach();
+
+ Queue queue = session.createQueue("myQueue");
+ final MessageProducer producer = session.createProducer(queue);
+
+ final int MSG_COUNT = 3;
+
+ for (int i = 0; i < MSG_COUNT; ++i) {
+ testPeer.expectTransferButDoNotRespond(new TransferPayloadCompositeMatcher());
+ }
+
+ TestJmsCompletionListener listener = new TestJmsCompletionListener(MSG_COUNT);
+ try {
+ for (int i = 0; i < MSG_COUNT; ++i) {
+ Message message = session.createTextMessage("content");
+ producer.send(message, listener);
+ }
+ } catch (JMSException e) {
+ LOG.warn("Caught unexpected error: {}", e.getMessage());
+ fail("No expected exception for this send.");
+ }
+
+ testPeer.waitForAllHandlersToComplete(2000);
+
+ assertFalse(listener.hasCompleted());
+
+ testPeer.expectSenderAttach();
+ testPeer.remotelyEndLastOpenedSession(true, 50, AmqpError.RESOURCE_DELETED, BREAD_CRUMB);
+
+ session.createProducer(queue);
+
+ // Verify the session gets marked closed
+ assertTrue(sessionClosed.await(5, TimeUnit.SECONDS), "Session closed callback didn't trigger");
+
+ try {
+ producer.getDeliveryMode();
+ fail("Expected ISE to be thrown due to being closed");
+ } catch (IllegalStateException jmsise) {
+ String errorMessage = jmsise.getCause().getMessage();
+ assertTrue(errorMessage.contains(AmqpError.RESOURCE_DELETED.toString()));
+ assertTrue(errorMessage.contains(BREAD_CRUMB));
+ }
+
+ assertTrue(listener.awaitCompletion(5, TimeUnit.SECONDS));
+ assertEquals(MSG_COUNT, listener.errorCount); // All sends should have been failed
+
+ // Try closing it explicitly, should effectively no-op in client.
+ // The test peer will throw during close if it sends anything.
+ session.close();
+
+ testPeer.expectClose();
+ connection.close();
+
+ testPeer.waitForAllHandlersToComplete(1000);
+ }
+ }
+
private boolean verifyConsumerClosure(final String BREAD_CRUMB, final MessageConsumer consumer) throws Exception {
return Wait.waitFor(new Wait.Condition() {
@Override
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/StreamMessageIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/StreamMessageIntegrationTest.java
index b7906e48..3b09dfc7 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/StreamMessageIntegrationTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/StreamMessageIntegrationTest.java
@@ -439,6 +439,7 @@ public class StreamMessageIntegrationTest extends QpidJmsTestCase {
try(TestAmqpPeer testPeer = new TestAmqpPeer();) {
JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer);
connection.setSendTimeout(15000);
+ connection.setSendTimeout(15);
testPeer.expectBegin();
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/TextMessageIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/TextMessageIntegrationTest.java
index 4653447c..1d907c6b 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/TextMessageIntegrationTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/TextMessageIntegrationTest.java
@@ -447,6 +447,7 @@ public class TextMessageIntegrationTest extends QpidJmsTestCase {
try(TestAmqpPeer testPeer = new TestAmqpPeer();) {
JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer);
connection.setSendTimeout(15000);
+ connection.setCloseTimeout(15);
testPeer.expectBegin();
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 9ae890f7..a6e350a8 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
@@ -4558,6 +4558,69 @@ public class FailoverIntegrationTest extends QpidJmsTestCase {
}
}
+ @Test
+ @Timeout(20)
+ public void testFailoverDoesFailPendingAsyncCompletionSend() throws Exception {
+ try (TestAmqpPeer originalPeer = new TestAmqpPeer();
+ TestAmqpPeer finalPeer = new TestAmqpPeer();) {
+
+ // 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);
+
+ originalPeer.expectSaslAnonymous();
+ originalPeer.expectOpen();
+ originalPeer.expectBegin();
+ originalPeer.expectBegin();
+ originalPeer.expectSenderAttach();
+ originalPeer.expectTransferButDoNotRespond(new TransferPayloadCompositeMatcher());
+ originalPeer.dropAfterLastHandler(15); // Wait for sender to get into wait state
+
+ // --- Post Failover Expectations of sender --- //
+ finalPeer.expectSaslAnonymous();
+ finalPeer.expectOpen();
+ finalPeer.expectBegin();
+ finalPeer.expectBegin();
+ finalPeer.expectSenderAttach();
+
+ final JmsConnection connection = establishAnonymousConnecton("failover.initialReconnectDelay=25", originalPeer, finalPeer);
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue("myQueue");
+
+ MessageProducer producer = session.createProducer(queue);
+
+ // Create and transfer a new message
+ String text = "myMessage";
+
+ TextMessage message = session.createTextMessage(text);
+ TestJmsCompletionListener listener = new TestJmsCompletionListener();
+
+ try {
+ producer.send(message, listener);
+ } catch (JMSException jmsEx) {
+ fail("Should not have failed the async completion send.");
+ }
+
+ // This should fire after reconnect without an error, if it fires with an error at
+ // any time then something is wrong.
+ assertTrue(listener.awaitCompletion(5, TimeUnit.SECONDS), "Did not get async callback");
+ assertNotNull(listener.exception, "Completion should have been due to error");
+ assertNotNull(listener.message);
+ assertTrue(listener.message instanceof TextMessage);
+
+ finalPeer.waitForAllHandlersToComplete(5000);
+ finalPeer.expectClose();
+
+ connection.close();
+
+ finalPeer.waitForAllHandlersToComplete(5000);
+ }
+ }
+
@Test
@Timeout(20)
public void testFailoverHandlesAnonymousFallbackWaitingForClose() throws Exception {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org