Skip to content

ARTEMIS-893 Fix FailoverTransactionTest.testWaitForMissingRedeliveries #925

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
gaohoward wants to merge 1 commit into from
Closed

ARTEMIS-893 Fix FailoverTransactionTest.testWaitForMissingRedeliveries #925

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
78 changes: 46 additions & 32 deletions ...t-tests/src/test/java/org/apache/activemq/transport/failover/FailoverTransactionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -385,6 +385,7 @@ public void run() {
LOG.info("Received: " + msg);
Assert.assertNull("no messges left dangling but got: " + msg, msg);
connection.close();
proxy.close();
}

@Test
Expand Down Expand Up @@ -857,6 +858,7 @@ public void testAutoRollbackWithMissingRedeliveries() throws Exception {
broker.stop();
broker = createBroker();
broker.start();

Assert.assertNotNull("should get rolledback message from original restarted broker", consumer.receive(20000));
} finally {
connection.close();
Expand All @@ -872,45 +874,57 @@ public void testWaitForMissingRedeliveries() throws Exception {
ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("failover:(" + url + ")?jms.consumerFailoverRedeliveryWaitPeriod=30000");
configureConnectionFactory(cf);
Connection connection = cf.createConnection();
connection.start();
final Session producerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
final Queue destination = producerSession.createQueue(QUEUE_NAME);
final Session consumerSession = connection.createSession(true, Session.SESSION_TRANSACTED);
MessageConsumer consumer = consumerSession.createConsumer(destination);
try {
connection.start();
final Session producerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
final Queue destination = producerSession.createQueue(QUEUE_NAME);
final Session consumerSession = connection.createSession(true, Session.SESSION_TRANSACTED);
MessageConsumer consumer = consumerSession.createConsumer(destination);

produceMessage(producerSession, destination);
Message msg = consumer.receive(20000);
if (msg == null) {
AutoFailTestSupport.dumpAllThreads("missing-");
}
Assert.assertNotNull("got message just produced", msg);
produceMessage(producerSession, destination);
Message msg = consumer.receive(20000);
if (msg == null) {
AutoFailTestSupport.dumpAllThreads("missing-");
}
Assert.assertNotNull("got message just produced", msg);

broker.stop();
broker = createBroker();
broker.start();
broker.stop();
broker = createBroker();
broker.start();

final CountDownLatch commitDone = new CountDownLatch(1);
// will block pending re-deliveries
new Thread() {
@Override
public void run() {
LOG.info("doing async commit...");
try {
consumerSession.commit();
commitDone.countDown();
} catch (JMSException ignored) {
final CountDownLatch commitDone = new CountDownLatch(1);
final CountDownLatch gotException = new CountDownLatch(1);
// will block pending re-deliveries
new Thread() {
@Override
public void run() {
LOG.info("doing async commit...");
try {
consumerSession.commit();
commitDone.countDown();
}
catch (JMSException ignored) {
System.out.println("--->err: got exfeption:");
ignored.printStackTrace();
gotException.countDown();
}
finally {
commitDone.countDown();
}
}
}
}.start();
}.start();

broker.stop();
broker = createBroker();
broker.start();
broker.stop();
broker = createBroker();
broker.start();

Assert.assertTrue("commit was successful", commitDone.await(30, TimeUnit.SECONDS));
Assert.assertTrue("commit was successful", commitDone.await(30, TimeUnit.SECONDS));
Assert.assertTrue("got exception on commit", gotException.await(30, TimeUnit.SECONDS));

Assert.assertNull("should not get committed message", consumer.receive(5000));
connection.close();
Assert.assertNotNull("should get failed committed message", consumer.receive(5000));
} finally {
connection.close();
}
}

@Test
Expand Down