mirror of https://github.com/apache/activemq.git
Update the tests so that they're not dependent on port 61616
git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1084797 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Timothy A. Bish
parent
5cf486c975
commit
883eed0e38
|
|
@ -52,55 +52,58 @@ import org.junit.After;
|
|||
import org.junit.Test;
|
||||
|
||||
public class FailoverConsumerOutstandingCommitTest {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(FailoverConsumerOutstandingCommitTest.class);
|
||||
private static final String QUEUE_NAME = "FailoverWithOutstandingCommit";
|
||||
private static final String MESSAGE_TEXT = "Test message ";
|
||||
private String url = "tcp://localhost:61616";
|
||||
final int prefetch = 10;
|
||||
BrokerService broker;
|
||||
|
||||
public void startCleanBroker() throws Exception {
|
||||
startBroker(true);
|
||||
}
|
||||
|
||||
@After
|
||||
public void stopBroker() throws Exception {
|
||||
if (broker != null) {
|
||||
broker.stop();
|
||||
}
|
||||
}
|
||||
|
||||
public void startBroker(boolean deleteAllMessagesOnStartup) throws Exception {
|
||||
broker = createBroker(deleteAllMessagesOnStartup);
|
||||
broker.start();
|
||||
}
|
||||
|
||||
public BrokerService createBroker(boolean deleteAllMessagesOnStartup) throws Exception {
|
||||
broker = new BrokerService();
|
||||
broker.addConnector(url);
|
||||
broker.setDeleteAllMessagesOnStartup(deleteAllMessagesOnStartup);
|
||||
PolicyMap policyMap = new PolicyMap();
|
||||
PolicyEntry defaultEntry = new PolicyEntry();
|
||||
|
||||
// optimizedDispatche and sync dispatch ensure that the dispatch happens
|
||||
// before the commit reply that the consumer.clearDispatchList is waiting for.
|
||||
defaultEntry.setOptimizedDispatch(true);
|
||||
private static final Logger LOG = LoggerFactory.getLogger(FailoverConsumerOutstandingCommitTest.class);
|
||||
private static final String QUEUE_NAME = "FailoverWithOutstandingCommit";
|
||||
private static final String MESSAGE_TEXT = "Test message ";
|
||||
private static final String TRANSPORT_URI = "tcp://localhost:0";
|
||||
private String url;
|
||||
final int prefetch = 10;
|
||||
BrokerService broker;
|
||||
|
||||
@After
|
||||
public void stopBroker() throws Exception {
|
||||
if (broker != null) {
|
||||
broker.stop();
|
||||
}
|
||||
}
|
||||
|
||||
public void startBroker(boolean deleteAllMessagesOnStartup) throws Exception {
|
||||
broker = createBroker(deleteAllMessagesOnStartup);
|
||||
broker.start();
|
||||
}
|
||||
|
||||
public BrokerService createBroker(boolean deleteAllMessagesOnStartup) throws Exception {
|
||||
return createBroker(deleteAllMessagesOnStartup, TRANSPORT_URI);
|
||||
}
|
||||
|
||||
public BrokerService createBroker(boolean deleteAllMessagesOnStartup, String bindAddress) throws Exception {
|
||||
broker = new BrokerService();
|
||||
broker.addConnector(bindAddress);
|
||||
broker.setDeleteAllMessagesOnStartup(deleteAllMessagesOnStartup);
|
||||
PolicyMap policyMap = new PolicyMap();
|
||||
PolicyEntry defaultEntry = new PolicyEntry();
|
||||
|
||||
// optimizedDispatche and sync dispatch ensure that the dispatch happens
|
||||
// before the commit reply that the consumer.clearDispatchList is waiting for.
|
||||
defaultEntry.setOptimizedDispatch(true);
|
||||
policyMap.setDefaultEntry(defaultEntry);
|
||||
broker.setDestinationPolicy(policyMap);
|
||||
|
||||
return broker;
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFailoverConsumerDups() throws Exception {
|
||||
doTestFailoverConsumerDups(true);
|
||||
}
|
||||
|
||||
public void doTestFailoverConsumerDups(final boolean watchTopicAdvisories) throws Exception {
|
||||
|
||||
url = broker.getTransportConnectors().get(0).getConnectUri().toString();
|
||||
|
||||
return broker;
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFailoverConsumerDups() throws Exception {
|
||||
doTestFailoverConsumerDups(true);
|
||||
}
|
||||
|
||||
public void doTestFailoverConsumerDups(final boolean watchTopicAdvisories) throws Exception {
|
||||
|
||||
broker = createBroker(true);
|
||||
|
||||
|
||||
broker.setPlugins(new BrokerPlugin[] {
|
||||
new BrokerPluginSupport() {
|
||||
@Override
|
||||
|
|
@ -108,7 +111,7 @@ public class FailoverConsumerOutstandingCommitTest {
|
|||
TransactionId xid, boolean onePhase) throws Exception {
|
||||
// so commit will hang as if reply is lost
|
||||
context.setDontSendReponse(true);
|
||||
Executors.newSingleThreadExecutor().execute(new Runnable() {
|
||||
Executors.newSingleThreadExecutor().execute(new Runnable() {
|
||||
public void run() {
|
||||
LOG.info("Stopping broker before commit...");
|
||||
try {
|
||||
|
|
@ -122,17 +125,17 @@ public class FailoverConsumerOutstandingCommitTest {
|
|||
}
|
||||
});
|
||||
broker.start();
|
||||
|
||||
|
||||
ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("failover:(" + url + ")");
|
||||
cf.setWatchTopicAdvisories(watchTopicAdvisories);
|
||||
cf.setDispatchAsync(false);
|
||||
|
||||
|
||||
final ActiveMQConnection connection = (ActiveMQConnection) cf.createConnection();
|
||||
connection.start();
|
||||
|
||||
|
||||
final Session producerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
|
||||
final Queue destination = producerSession.createQueue(QUEUE_NAME + "?consumer.prefetchSize=" + prefetch);
|
||||
|
||||
|
||||
final Session consumerSession = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
|
||||
|
||||
|
||||
|
|
@ -144,9 +147,9 @@ public class FailoverConsumerOutstandingCommitTest {
|
|||
|
||||
public void onMessage(Message message) {
|
||||
LOG.info("consume one and commit");
|
||||
|
||||
|
||||
assertNotNull("got message", message);
|
||||
|
||||
|
||||
try {
|
||||
consumerSession.commit();
|
||||
} catch (JMSException e) {
|
||||
|
|
@ -157,7 +160,7 @@ public class FailoverConsumerOutstandingCommitTest {
|
|||
LOG.info("done commit");
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
// may block if broker shutodwn happens quickly
|
||||
Executors.newSingleThreadExecutor().execute(new Runnable() {
|
||||
public void run() {
|
||||
|
|
@ -171,15 +174,15 @@ public class FailoverConsumerOutstandingCommitTest {
|
|||
LOG.info("producer done");
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
// will be stopped by the plugin
|
||||
broker.waitUntilStopped();
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, url);
|
||||
broker.start();
|
||||
|
||||
assertTrue("consumer added through failover", commitDoneLatch.await(20, TimeUnit.SECONDS));
|
||||
assertTrue("another message was recieved after failover", messagesReceived.await(20, TimeUnit.SECONDS));
|
||||
|
||||
|
||||
connection.close();
|
||||
}
|
||||
|
||||
|
|
@ -187,12 +190,12 @@ public class FailoverConsumerOutstandingCommitTest {
|
|||
public void TestFailoverConsumerOutstandingSendTxIncomplete() throws Exception {
|
||||
doTestFailoverConsumerOutstandingSendTx(false);
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void TestFailoverConsumerOutstandingSendTxComplete() throws Exception {
|
||||
doTestFailoverConsumerOutstandingSendTx(true);
|
||||
}
|
||||
|
||||
|
||||
public void doTestFailoverConsumerOutstandingSendTx(final boolean doActualBrokerCommit) throws Exception {
|
||||
final boolean watchTopicAdvisories = true;
|
||||
broker = createBroker(true);
|
||||
|
|
@ -233,7 +236,7 @@ public class FailoverConsumerOutstandingCommitTest {
|
|||
final Session producerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
|
||||
final Queue destination = producerSession.createQueue(QUEUE_NAME
|
||||
+ "?consumer.prefetchSize=" + prefetch);
|
||||
|
||||
|
||||
final Queue signalDestination = producerSession.createQueue(QUEUE_NAME + ".signal"
|
||||
+ "?consumer.prefetchSize=" + prefetch);
|
||||
|
||||
|
|
@ -280,7 +283,7 @@ public class FailoverConsumerOutstandingCommitTest {
|
|||
|
||||
// will be stopped by the plugin
|
||||
broker.waitUntilStopped();
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, url);
|
||||
broker.start();
|
||||
|
||||
assertTrue("commit done through failover", commitDoneLatch.await(20, TimeUnit.SECONDS));
|
||||
|
|
@ -291,8 +294,8 @@ public class FailoverConsumerOutstandingCommitTest {
|
|||
assertEquals("get message 0 second", MESSAGE_TEXT + "0", receivedMessages.get(1).getText());
|
||||
assertTrue("another message was received", messagesReceived.await(20, TimeUnit.SECONDS));
|
||||
assertEquals("get message 1 eventually", MESSAGE_TEXT + "1", receivedMessages.get(2).getText());
|
||||
|
||||
|
||||
|
||||
|
||||
connection.close();
|
||||
}
|
||||
|
||||
|
|
@ -312,28 +315,28 @@ public class FailoverConsumerOutstandingCommitTest {
|
|||
final Session consumerSession = connection.createSession(true, Session.SESSION_TRANSACTED);
|
||||
final MessageConsumer testConsumer = consumerSession.createConsumer(destination);
|
||||
assertNull("no message yet", testConsumer.receiveNoWait());
|
||||
|
||||
|
||||
produceMessage(producerSession, destination, 1);
|
||||
producerSession.close();
|
||||
|
||||
// consume then rollback after restart
|
||||
Message msg = testConsumer.receive(5000);
|
||||
assertNotNull(msg);
|
||||
|
||||
|
||||
// restart with outstanding delivered message
|
||||
broker.stop();
|
||||
broker.waitUntilStopped();
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, url);
|
||||
broker.start();
|
||||
|
||||
|
||||
consumerSession.rollback();
|
||||
|
||||
|
||||
// receive again
|
||||
msg = testConsumer.receive(10000);
|
||||
assertNotNull("got message again after rollback", msg);
|
||||
|
||||
consumerSession.commit();
|
||||
|
||||
|
||||
// close before sweep
|
||||
consumerSession.close();
|
||||
msg = receiveMessage(cf, destination);
|
||||
|
|
|
|||
|
|
@ -51,51 +51,55 @@ import org.junit.Test;
|
|||
|
||||
// see https://issues.apache.org/activemq/browse/AMQ-2573
|
||||
public class FailoverConsumerUnconsumedTest {
|
||||
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(FailoverConsumerUnconsumedTest.class);
|
||||
private static final String QUEUE_NAME = "FailoverWithUnconsumed";
|
||||
private String url = "tcp://localhost:61616";
|
||||
final int prefetch = 10;
|
||||
BrokerService broker;
|
||||
|
||||
public void startCleanBroker() throws Exception {
|
||||
startBroker(true);
|
||||
}
|
||||
|
||||
@After
|
||||
public void stopBroker() throws Exception {
|
||||
if (broker != null) {
|
||||
broker.stop();
|
||||
}
|
||||
}
|
||||
|
||||
public void startBroker(boolean deleteAllMessagesOnStartup) throws Exception {
|
||||
broker = createBroker(deleteAllMessagesOnStartup);
|
||||
private static final String QUEUE_NAME = "FailoverWithUnconsumed";
|
||||
private static final String TRANSPORT_URI = "tcp://localhost:0";
|
||||
private String url;
|
||||
final int prefetch = 10;
|
||||
BrokerService broker;
|
||||
|
||||
@After
|
||||
public void stopBroker() throws Exception {
|
||||
if (broker != null) {
|
||||
broker.stop();
|
||||
}
|
||||
}
|
||||
|
||||
public void startBroker(boolean deleteAllMessagesOnStartup) throws Exception {
|
||||
broker = createBroker(deleteAllMessagesOnStartup);
|
||||
broker.start();
|
||||
}
|
||||
}
|
||||
|
||||
public BrokerService createBroker(boolean deleteAllMessagesOnStartup) throws Exception {
|
||||
broker = new BrokerService();
|
||||
broker.addConnector(url);
|
||||
broker.setDeleteAllMessagesOnStartup(deleteAllMessagesOnStartup);
|
||||
return broker;
|
||||
}
|
||||
public BrokerService createBroker(boolean deleteAllMessagesOnStartup) throws Exception {
|
||||
return createBroker(deleteAllMessagesOnStartup, TRANSPORT_URI);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFailoverConsumerDups() throws Exception {
|
||||
doTestFailoverConsumerDups(true);
|
||||
}
|
||||
|
||||
@Test
|
||||
public BrokerService createBroker(boolean deleteAllMessagesOnStartup, String bindAddress) throws Exception {
|
||||
broker = new BrokerService();
|
||||
broker.addConnector(bindAddress);
|
||||
broker.setDeleteAllMessagesOnStartup(deleteAllMessagesOnStartup);
|
||||
|
||||
this.url = broker.getTransportConnectors().get(0).getConnectUri().toString();
|
||||
|
||||
return broker;
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFailoverConsumerDups() throws Exception {
|
||||
doTestFailoverConsumerDups(true);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFailoverConsumerDupsNoAdvisoryWatch() throws Exception {
|
||||
doTestFailoverConsumerDups(false);
|
||||
}
|
||||
|
||||
public void doTestFailoverConsumerDups(final boolean watchTopicAdvisories) throws Exception {
|
||||
|
||||
final int maxConsumers = 4;
|
||||
|
||||
public void doTestFailoverConsumerDups(final boolean watchTopicAdvisories) throws Exception {
|
||||
|
||||
final int maxConsumers = 4;
|
||||
broker = createBroker(true);
|
||||
|
||||
|
||||
broker.setPlugins(new BrokerPlugin[] {
|
||||
new BrokerPluginSupport() {
|
||||
int consumerCount;
|
||||
|
|
@ -106,7 +110,7 @@ public class FailoverConsumerUnconsumedTest {
|
|||
final ConsumerInfo info) throws Exception {
|
||||
if (++consumerCount == maxConsumers + (watchTopicAdvisories ? 1:0)) {
|
||||
context.setDontSendReponse(true);
|
||||
Executors.newSingleThreadExecutor().execute(new Runnable() {
|
||||
Executors.newSingleThreadExecutor().execute(new Runnable() {
|
||||
public void run() {
|
||||
LOG.info("Stopping broker on consumer: " + info.getConsumerId());
|
||||
try {
|
||||
|
|
@ -122,13 +126,13 @@ public class FailoverConsumerUnconsumedTest {
|
|||
}
|
||||
});
|
||||
broker.start();
|
||||
|
||||
|
||||
ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("failover:(" + url + ")");
|
||||
cf.setWatchTopicAdvisories(watchTopicAdvisories);
|
||||
|
||||
|
||||
final ActiveMQConnection connection = (ActiveMQConnection) cf.createConnection();
|
||||
connection.start();
|
||||
|
||||
|
||||
final Session consumerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
|
||||
final Queue destination = consumerSession.createQueue(QUEUE_NAME + "?jms.consumer.prefetch=" + prefetch);
|
||||
|
||||
|
|
@ -136,9 +140,9 @@ public class FailoverConsumerUnconsumedTest {
|
|||
for (int i=0; i<maxConsumers -1; i++) {
|
||||
testConsumers.add(new TestConsumer(consumerSession, destination, connection));
|
||||
}
|
||||
|
||||
|
||||
produceMessage(consumerSession, destination, maxConsumers * prefetch);
|
||||
|
||||
|
||||
assertTrue("add messages are dispatched", Wait.waitFor(new Wait.Condition() {
|
||||
public boolean isSatisified() throws Exception {
|
||||
int totalUnconsumed = 0;
|
||||
|
|
@ -146,14 +150,14 @@ public class FailoverConsumerUnconsumedTest {
|
|||
long unconsumed = testConsumer.unconsumedSize();
|
||||
LOG.info(testConsumer.getConsumerId() + " unconsumed: " + unconsumed);
|
||||
totalUnconsumed += unconsumed;
|
||||
}
|
||||
}
|
||||
return totalUnconsumed == (maxConsumers-1) * prefetch;
|
||||
}
|
||||
}));
|
||||
|
||||
|
||||
final CountDownLatch commitDoneLatch = new CountDownLatch(1);
|
||||
|
||||
Executors.newSingleThreadExecutor().execute(new Runnable() {
|
||||
|
||||
Executors.newSingleThreadExecutor().execute(new Runnable() {
|
||||
public void run() {
|
||||
try {
|
||||
LOG.info("add last consumer...");
|
||||
|
|
@ -165,7 +169,7 @@ public class FailoverConsumerUnconsumedTest {
|
|||
}
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
// will be stopped by the plugin
|
||||
broker.waitUntilStopped();
|
||||
|
||||
|
|
@ -182,11 +186,11 @@ public class FailoverConsumerUnconsumedTest {
|
|||
}
|
||||
}));
|
||||
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, this.url);
|
||||
broker.start();
|
||||
|
||||
assertTrue("consumer added through failover", commitDoneLatch.await(30, TimeUnit.SECONDS));
|
||||
|
||||
|
||||
// each should again get prefetch messages - all unconsumed deliveries should be rolledback
|
||||
assertTrue("after start all messages are re dispatched", Wait.waitFor(new Wait.Condition() {
|
||||
public boolean isSatisified() throws Exception {
|
||||
|
|
@ -195,14 +199,14 @@ public class FailoverConsumerUnconsumedTest {
|
|||
long unconsumed = testConsumer.unconsumedSize();
|
||||
LOG.info(testConsumer.getConsumerId() + " after restart: unconsumed: " + unconsumed);
|
||||
totalUnconsumed += unconsumed;
|
||||
}
|
||||
}
|
||||
return totalUnconsumed == (maxConsumers) * prefetch;
|
||||
}
|
||||
}));
|
||||
|
||||
|
||||
connection.close();
|
||||
}
|
||||
|
||||
|
||||
private void produceMessage(final Session producerSession, Queue destination, long count)
|
||||
throws JMSException {
|
||||
MessageProducer producer = producerSession.createProducer(destination);
|
||||
|
|
@ -212,22 +216,22 @@ public class FailoverConsumerUnconsumedTest {
|
|||
}
|
||||
producer.close();
|
||||
}
|
||||
|
||||
|
||||
// allow access to unconsumedMessages
|
||||
class TestConsumer extends ActiveMQMessageConsumer {
|
||||
|
||||
|
||||
TestConsumer(Session consumerSession, Destination destination, ActiveMQConnection connection) throws Exception {
|
||||
super((ActiveMQSession) consumerSession,
|
||||
new ConsumerId(new SessionId(connection.getConnectionInfo().getConnectionId(),1), nextGen()),
|
||||
super((ActiveMQSession) consumerSession,
|
||||
new ConsumerId(new SessionId(connection.getConnectionInfo().getConnectionId(),1), nextGen()),
|
||||
ActiveMQMessageTransformation.transformDestination(destination), null, "",
|
||||
prefetch, -1, false, false, true, null);
|
||||
}
|
||||
|
||||
|
||||
public int unconsumedSize() {
|
||||
return unconsumedMessages.size();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
static long idGen = 100;
|
||||
private static long nextGen() {
|
||||
idGen -=5;
|
||||
|
|
|
|||
|
|
@ -48,14 +48,11 @@ public class FailoverPrefetchZeroTest {
|
|||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(FailoverPrefetchZeroTest.class);
|
||||
private static final String QUEUE_NAME = "FailoverPrefetchZero";
|
||||
private String url = "tcp://localhost:61616";
|
||||
private static final String TRANSPORT_URI = "tcp://localhost:0";
|
||||
private String url;
|
||||
final int prefetch = 0;
|
||||
BrokerService broker;
|
||||
|
||||
public void startCleanBroker() throws Exception {
|
||||
startBroker(true);
|
||||
}
|
||||
|
||||
@After
|
||||
public void stopBroker() throws Exception {
|
||||
if (broker != null) {
|
||||
|
|
@ -69,9 +66,16 @@ public class FailoverPrefetchZeroTest {
|
|||
}
|
||||
|
||||
public BrokerService createBroker(boolean deleteAllMessagesOnStartup) throws Exception {
|
||||
return createBroker(deleteAllMessagesOnStartup, TRANSPORT_URI);
|
||||
}
|
||||
|
||||
public BrokerService createBroker(boolean deleteAllMessagesOnStartup, String bindAddress) throws Exception {
|
||||
broker = new BrokerService();
|
||||
broker.addConnector(url);
|
||||
broker.addConnector(bindAddress);
|
||||
broker.setDeleteAllMessagesOnStartup(deleteAllMessagesOnStartup);
|
||||
|
||||
url = broker.getTransportConnectors().get(0).getConnectUri().toString();
|
||||
|
||||
return broker;
|
||||
}
|
||||
|
||||
|
|
@ -135,7 +139,7 @@ public class FailoverPrefetchZeroTest {
|
|||
// will be stopped by the plugin
|
||||
assertTrue("pull completed on broker", pullDone.await(30, TimeUnit.SECONDS));
|
||||
broker.waitUntilStopped();
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, url);
|
||||
broker.start();
|
||||
|
||||
assertTrue("receive completed through failover", receiveDone.await(30, TimeUnit.SECONDS));
|
||||
|
|
|
|||
|
|
@ -60,17 +60,14 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(FailoverTransactionTest.class);
|
||||
private static final String QUEUE_NAME = "FailoverWithTx";
|
||||
private String url = "tcp://localhost:61616";
|
||||
private static final String TRANSPORT_URI = "tcp://localhost:0";
|
||||
private String url;
|
||||
BrokerService broker;
|
||||
|
||||
public static Test suite() {
|
||||
return suite(FailoverTransactionTest.class);
|
||||
}
|
||||
|
||||
public void startCleanBroker() throws Exception {
|
||||
startBroker(true);
|
||||
}
|
||||
|
||||
public void setUp() throws Exception {
|
||||
super.setMaxTestTime(20 * 60 * 1000); // some boxes can be real slow
|
||||
super.setAutoFail(true);
|
||||
|
|
@ -87,17 +84,33 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
}
|
||||
}
|
||||
|
||||
private void startCleanBroker() throws Exception {
|
||||
startBroker(true);
|
||||
}
|
||||
|
||||
public void startBroker(boolean deleteAllMessagesOnStartup) throws Exception {
|
||||
broker = createBroker(deleteAllMessagesOnStartup);
|
||||
broker.start();
|
||||
}
|
||||
|
||||
public void startBroker(boolean deleteAllMessagesOnStartup, String bindAddress) throws Exception {
|
||||
broker = createBroker(deleteAllMessagesOnStartup, bindAddress);
|
||||
broker.start();
|
||||
}
|
||||
|
||||
public BrokerService createBroker(boolean deleteAllMessagesOnStartup) throws Exception {
|
||||
return createBroker(deleteAllMessagesOnStartup, TRANSPORT_URI);
|
||||
}
|
||||
|
||||
public BrokerService createBroker(boolean deleteAllMessagesOnStartup, String bindAddress) throws Exception {
|
||||
broker = new BrokerService();
|
||||
broker.setUseJmx(false);
|
||||
broker.setAdvisorySupport(false);
|
||||
broker.addConnector(url);
|
||||
broker.addConnector(bindAddress);
|
||||
broker.setDeleteAllMessagesOnStartup(deleteAllMessagesOnStartup);
|
||||
|
||||
url = broker.getTransportConnectors().get(0).getConnectUri().toString();
|
||||
|
||||
return broker;
|
||||
}
|
||||
|
||||
|
|
@ -114,7 +127,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
|
||||
// restart to force failover and connection state recovery before the commit
|
||||
broker.stop();
|
||||
startBroker(false);
|
||||
startBroker(false, url);
|
||||
|
||||
session.commit();
|
||||
assertNotNull("we got the message", consumer.receive(20000));
|
||||
|
|
@ -182,7 +195,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
|
||||
// will be stopped by the plugin
|
||||
broker.waitUntilStopped();
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, url);
|
||||
setDefaultPersistenceAdapter(broker);
|
||||
broker.start();
|
||||
|
||||
|
|
@ -202,7 +215,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
broker.waitUntilStopped();
|
||||
|
||||
LOG.info("Checking for remaining/hung messages..");
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, url);
|
||||
setDefaultPersistenceAdapter(broker);
|
||||
broker.start();
|
||||
|
||||
|
|
@ -285,7 +298,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
|
||||
// will be stopped by the plugin
|
||||
broker.waitUntilStopped();
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, url);
|
||||
setDefaultPersistenceAdapter(broker);
|
||||
LOG.info("restarting....");
|
||||
broker.start();
|
||||
|
|
@ -309,7 +322,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
broker.waitUntilStopped();
|
||||
|
||||
LOG.info("Checking for remaining/hung messages with second restart..");
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, url);
|
||||
setDefaultPersistenceAdapter(broker);
|
||||
broker.start();
|
||||
|
||||
|
|
@ -430,7 +443,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
broker.waitUntilStopped();
|
||||
|
||||
LOG.info("Checking for remaining/hung messages with restart..");
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, url);
|
||||
setDefaultPersistenceAdapter(broker);
|
||||
broker.start();
|
||||
|
||||
|
|
@ -462,7 +475,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
|
||||
// restart to force failover and connection state recovery before the commit
|
||||
broker.stop();
|
||||
startBroker(false);
|
||||
startBroker(false, url);
|
||||
|
||||
session.commit();
|
||||
|
||||
|
|
@ -493,7 +506,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
|
||||
// restart to force failover and connection state recovery before the commit
|
||||
broker.stop();
|
||||
startBroker(false);
|
||||
startBroker(false, url);
|
||||
|
||||
session.commit();
|
||||
for (int i = 0; i < count; i++) {
|
||||
|
|
@ -543,7 +556,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
|
||||
// restart to force failover and connection state recovery before the commit
|
||||
broker.stop();
|
||||
startBroker(false);
|
||||
startBroker(false, url);
|
||||
|
||||
session.commit();
|
||||
for (int i = 0; i < count - 1; i++) {
|
||||
|
|
@ -671,7 +684,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
|
||||
// will be stopped by the plugin
|
||||
broker.waitUntilStopped();
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, url);
|
||||
setDefaultPersistenceAdapter(broker);
|
||||
broker.start();
|
||||
|
||||
|
|
@ -708,7 +721,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
broker.waitUntilStopped();
|
||||
|
||||
LOG.info("Checking for remaining/hung messages..");
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, url);
|
||||
setDefaultPersistenceAdapter(broker);
|
||||
broker.start();
|
||||
|
||||
|
|
@ -744,7 +757,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
assertNotNull(msg);
|
||||
|
||||
broker.stop();
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, url);
|
||||
// use empty jdbc store so that default wait(0) for redeliveries will timeout after failover
|
||||
setPersistenceAdapter(broker, PersistenceAdapterChoice.JDBC);
|
||||
broker.start();
|
||||
|
|
@ -756,7 +769,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
}
|
||||
|
||||
broker.stop();
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, url);
|
||||
broker.start();
|
||||
|
||||
assertNotNull("should get rolledback message from original restarted broker", consumer.receive(20000));
|
||||
|
|
@ -784,7 +797,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
assertNotNull("got message just produced", msg);
|
||||
|
||||
broker.stop();
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, url);
|
||||
// use empty jdbc store so that wait for re-deliveries occur when failover resumes
|
||||
setPersistenceAdapter(broker, PersistenceAdapterChoice.JDBC);
|
||||
broker.start();
|
||||
|
|
@ -803,7 +816,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
});
|
||||
|
||||
broker.stop();
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, url);
|
||||
broker.start();
|
||||
|
||||
assertTrue("commit was successful", commitDone.await(30, TimeUnit.SECONDS));
|
||||
|
|
@ -836,7 +849,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
MessageConsumer consumer2 = consumerSession.createConsumer(consumerSession.createQueue(QUEUE_NAME + "?consumer.prefetchSize=1"));
|
||||
|
||||
broker.stop();
|
||||
broker = createBroker(false);
|
||||
broker = createBroker(false, url);
|
||||
broker.start();
|
||||
|
||||
final CountDownLatch commitDone = new CountDownLatch(1);
|
||||
|
|
@ -851,7 +864,7 @@ public class FailoverTransactionTest extends TestSupport {
|
|||
try {
|
||||
consumerSession.commit();
|
||||
} catch (JMSException ex) {
|
||||
exceptions.add(ex);
|
||||
exceptions.add(ex);
|
||||
} finally {
|
||||
commitDone.countDown();
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue