Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

further tests and fixes related to failover, this time transactions and topics. prefetch and maxPageSize are relevant in the transacted. With redeliveries the prefetch needs to be less than half the transaction size and the maxPageSize needs to exceed the transaction span. more tests in AMQ-2149|https://issues.apache.org/activemq/browse/AMQ-2149 and some amendments to the fixes for that issue 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@761597 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Tully 2009-04-03 10:16:28 +00:00
parent f8ef7ff31f
commit e45bb06907
13 changed files with 224 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
View File

@ -1838,7 +1838,9 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
}
for (Iterator<ActiveMQSession> i = this.sessions.iterator(); i.hasNext();) {
ActiveMQSession s = i.next();
s.deliverAcks();
// deliverAcks at this point is too early as acks can arrive at the broker
// before redispatch of messages and hence be out or order
s.transportResumed();
}
}

67
activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
View File

@ -131,7 +131,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
private MessageAck pendingAck;
private long lastDeliveredSequenceId;
private IOException failureError;
/**
@ -439,8 +439,8 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
timeout = Math.max(deadline - System.currentTimeMillis(), 0);
}
} else {
if (LOG.isDebugEnabled()) {
LOG.debug(getConsumerId() + " received message: " + md);
if (LOG.isTraceEnabled()) {
LOG.trace(getConsumerId() + " received message: " + md);
}
return md;
}
@ -639,18 +639,20 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
MessageAck ack = null;
if (deliveryingAcknowledgements.compareAndSet(false, true)) {
if (session.isAutoAcknowledge()) {
synchronized(deliveredMessages) {
ack = makeAckForAllDeliveredMessages(MessageAck.STANDARD_ACK_TYPE);
if (ack != null) {
deliveredMessages.clear();
ackCounter = 0;
synchronized(deliveredMessages) {
ack = makeAckForAllDeliveredMessages(MessageAck.STANDARD_ACK_TYPE);
if (ack != null) {
deliveredMessages.clear();
ackCounter = 0;
}
}
} else if (pendingAck != null && pendingAck.isStandardAck()) {
ack = pendingAck;
pendingAck = null;
}
if (ack != null) {
final MessageAck ackToSend = ack;
if (executorService == null) {
executorService = Executors.newSingleThreadExecutor();
}
@ -840,8 +842,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
private void ackLater(MessageDispatch md, byte ackType) throws JMSException {
// Don't acknowledge now, but we may need to let the broker know the
// consumer got the message
// to expand the pre-fetch window
// consumer got the message to expand the pre-fetch window
if (session.getTransacted()) {
session.doStartTransaction();
if (!synchronizationRegistered) {
@ -865,19 +866,30 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
}
}
// The delivered message list is only needed for the recover method
// which is only used with client ack.
deliveredCounter++;
MessageAck oldPendingAck = pendingAck;
pendingAck = new MessageAck(md, ackType, deliveredCounter);
pendingAck.setTransactionId(session.getTransactionContext().getTransactionId());
if( oldPendingAck==null ) {
pendingAck.setFirstMessageId(pendingAck.getLastMessageId());
} else {
} else if ( oldPendingAck.getAckType() == pendingAck.getAckType() ) {
pendingAck.setFirstMessageId(oldPendingAck.getFirstMessageId());
} else {
// old pending ack being superseded by ack of another type, if is is not a delivered
// ack and hence important, send it now so it is not lost.
if ( !oldPendingAck.isDeliveredAck()) {
if (LOG.isDebugEnabled()) {
LOG.debug("Sending old pending ack " + oldPendingAck + ", new pending: " + pendingAck);
}
session.sendAck(oldPendingAck);
} else {
if (LOG.isDebugEnabled()) {
LOG.debug("dropping old pending ack " + oldPendingAck + ", new pending: " + pendingAck);
}
}
}
pendingAck.setTransactionId(session.getTransactionContext().getTransactionId());
if ((0.5 * info.getPrefetchSize()) <= (deliveredCounter - additionalWindowSize)) {
session.sendAck(pendingAck);
pendingAck=null;
@ -910,14 +922,14 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
}
session.sendAck(ack);
pendingAck = null;
// Adjust the counters
deliveredCounter -= deliveredMessages.size();
deliveredCounter = Math.max(0, deliveredCounter - deliveredMessages.size());
additionalWindowSize = Math.max(0, additionalWindowSize - deliveredMessages.size());
if (!session.getTransacted()) {
if (!session.getTransacted()) {
deliveredMessages.clear();
}
}
}
}
@ -1073,9 +1085,12 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
} else {
// ignore duplicate
if (LOG.isDebugEnabled()) {
LOG.debug(getConsumerId() + " Ignoring Duplicate: " + md.getMessage());
LOG.debug(getConsumerId() + " ignoring duplicate: " + md.getMessage());
}
acknowledge(md);
// in a transaction ack delivery of duplicates to ensure prefetch extension kicks in.
// the normal ack will happen in the transaction.
ackLater(md, session.isTransacted() ?
MessageAck.DELIVERED_ACK_TYPE : MessageAck.STANDARD_ACK_TYPE);
}
}
}
@ -1144,6 +1159,13 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
return lastDeliveredSequenceId;
}
// on resumption re deliveries will percolate acks in their own good time
public void transportResumed() {
pendingAck = null;
additionalWindowSize = 0;
deliveredCounter = 0;
}
public IOException getFailureError() {
return failureError;
}
@ -1151,5 +1173,4 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
public void setFailureError(IOException failureError) {
this.failureError = failureError;
}
}

11
activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java
View File

@ -1669,8 +1669,8 @@ public class ActiveMQSession implements Session, QueueSession, TopicSession, Sta
msg.setConnection(connection);
msg.onSend();
msg.setProducerId(msg.getMessageId().getProducerId());
if (this.debug) {
LOG.debug(getSessionId() + " sending message: " + msg);
if (LOG.isTraceEnabled()) {
LOG.trace(getSessionId() + " sending message: " + msg);
}
if (sendTimeout <= 0 && !msg.isResponseRequired() && !connection.isAlwaysSyncSend() && (!msg.isPersistent() || connection.isUseAsyncSend() || txid != null)) {
this.connection.asyncSendPacket(msg);
@ -1963,4 +1963,11 @@ public class ActiveMQSession implements Session, QueueSession, TopicSession, Sta
}
}
public void transportResumed() {
for (Iterator<ActiveMQMessageConsumer> iter = consumers.iterator(); iter.hasNext();) {
ActiveMQMessageConsumer consumer = iter.next();
consumer.transportResumed();
}
}
}

15
activemq-core/src/main/java/org/apache/activemq/TransactionContext.java
View File

@ -211,10 +211,11 @@ public class TransactionContext implements XAResource {
if (localTransactionEventListener != null) {
localTransactionEventListener.beginEvent();
}
if (LOG.isDebugEnabled()) {
LOG.debug("Begin:" + transactionId);
}
}
if (LOG.isDebugEnabled()) {
LOG.debug("Being:" + transactionId);
}
}
/**
@ -234,7 +235,9 @@ public class TransactionContext implements XAResource {
beforeEnd();
if (transactionId != null) {
if (LOG.isDebugEnabled()) {
LOG.debug("Rollback:" + transactionId);
LOG.debug("Rollback: " + transactionId
+ " syncCount: "
+ (synchronizations != null ? synchronizations.size() : 0));
}
TransactionInfo info = new TransactionInfo(getConnectionId(), transactionId, TransactionInfo.ROLLBACK);
@ -268,7 +271,9 @@ public class TransactionContext implements XAResource {
// Only send commit if the transaction was started.
if (transactionId != null) {
if (LOG.isDebugEnabled()) {
LOG.debug("Commit:" + transactionId);
LOG.debug("Commit: " + transactionId
+ " syncCount: "
+ (synchronizations != null ? synchronizations.size() : 0));
}
TransactionInfo info = new TransactionInfo(getConnectionId(), transactionId, TransactionInfo.COMMIT_ONE_PHASE);

2
activemq-core/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java
View File

@ -199,7 +199,7 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
protected void doAddRecoveredMessage(MessageReference message) throws Exception {
synchronized(pending) {
pending.addRecoveredMessage(message);
pending.addRecoveredMessage(message);
}
}

8
activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
View File

@ -393,10 +393,8 @@ public abstract class PrefetchSubscription extends AbstractSubscription {
}
}
}
if (callDispatchMatched && destination != null) {
// if (destination.isLazyDispatch()) {
destination.wakeup();
// }
if (callDispatchMatched && destination != null) {
destination.wakeup();
dispatchPending();
} else {
if (isSlave()) {
@ -661,7 +659,7 @@ public abstract class PrefetchSubscription extends AbstractSubscription {
if (node.getRegionDestination() != null) {
if (node != QueueMessageReference.NULL_MESSAGE) {
node.getRegionDestination().getDestinationStatistics().getDispatched().increment();
node.getRegionDestination().getDestinationStatistics().getInflight().increment();
node.getRegionDestination().getDestinationStatistics().getInflight().increment();
}
}
if (LOG.isTraceEnabled()) {

7
activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
View File

@ -1258,6 +1258,13 @@ public class Queue extends BaseDestination implements Task, UsageListener {
dispatchLock.lock();
try{
int toPageIn = getMaxPageSize() + Math.max(0, (int)destinationStatistics.getInflight().getCount()) - pagedInMessages.size();
if (LOG.isDebugEnabled()) {
LOG.debug(destination.getPhysicalName() + " toPageIn: " + toPageIn + ", Inflight: "
+ destinationStatistics.getInflight().getCount()
+ ", pagedInMessages.size " + pagedInMessages.size());
}
toPageIn = Math.max(0, Math.min(toPageIn, getMaxPageSize()));
if (isLazyDispatch()&& !force) {
// Only page in the minimum number of messages which can be dispatched immediately.

26
activemq-core/src/main/java/org/apache/activemq/state/ConnectionStateTracker.java
View File

@ -39,6 +39,8 @@ import org.apache.activemq.command.SessionInfo;
import org.apache.activemq.command.TransactionInfo;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.util.IOExceptionSupport;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* Tracks the state of a connection so a newly established transport can be
@ -47,6 +49,7 @@ import org.apache.activemq.util.IOExceptionSupport;
* @version $Revision$
*/
public class ConnectionStateTracker extends CommandVisitorAdapter {
private static final Log LOG = LogFactory.getLog(ConnectionStateTracker.class);
private static final Tracked TRACKED_RESPONSE_MARKER = new Tracked(null);
@ -135,8 +138,14 @@ public class ConnectionStateTracker extends CommandVisitorAdapter {
private void restoreTransactions(Transport transport, ConnectionState connectionState) throws IOException {
for (Iterator iter = connectionState.getTransactionStates().iterator(); iter.hasNext();) {
TransactionState transactionState = (TransactionState)iter.next();
if (LOG.isDebugEnabled()) {
LOG.debug("tx: " + transactionState.getId());
}
for (Iterator iterator = transactionState.getCommands().iterator(); iterator.hasNext();) {
Command command = (Command)iterator.next();
if (LOG.isDebugEnabled()) {
LOG.debug("tx replay: " + command);
}
transport.oneway(command);
}
}
@ -359,23 +368,6 @@ public class ConnectionStateTracker extends CommandVisitorAdapter {
return null;
}
public Response processMessageAck(MessageAck ack) {
if (trackTransactions && ack != null && ack.getTransactionId() != null) {
ConnectionId connectionId = ack.getConsumerId().getParentId().getParentId();
if (connectionId != null) {
ConnectionState cs = connectionStates.get(connectionId);
if (cs != null) {
TransactionState transactionState = cs.getTransactionState(ack.getTransactionId());
if (transactionState != null) {
transactionState.addCommand(ack);
}
}
}
return TRACKED_RESPONSE_MARKER;
}
return null;
}
public Response processBeginTransaction(TransactionInfo info) {
if (trackTransactions && info != null && info.getTransactionId() != null) {
ConnectionId connectionId = info.getConnectionId();

19
activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaTopicReferenceStore.java
View File

@ -19,8 +19,10 @@ package org.apache.activemq.store.kahadaptor;
import java.io.IOException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.activemq.broker.ConnectionContext;
@ -101,8 +103,8 @@ public class KahaTopicReferenceStore extends KahaReferenceStore implements Topic
ref.setMessageId(messageId);
container.add(ref);
}
if (LOG.isDebugEnabled()) {
LOG.debug(destination.getPhysicalName() + " add reference: " + messageId);
if (LOG.isTraceEnabled()) {
LOG.trace(destination.getPhysicalName() + " add reference: " + messageId);
}
}
} finally {
@ -173,11 +175,11 @@ public class KahaTopicReferenceStore extends KahaReferenceStore implements Topic
ackContainer.update(entry,tsa);
}
}
if (LOG.isDebugEnabled()) {
LOG.debug(destination.getPhysicalName() + " remove: " + messageId);
if (LOG.isTraceEnabled()) {
LOG.trace(destination.getPhysicalName() + " remove: " + messageId);
}
}else{
if (ackContainer.isEmpty() || isUnreferencedBySubscribers(subscriberMessages, messageId)) {
if (ackContainer.isEmpty() || subscriberMessages.size() == 1 || isUnreferencedBySubscribers(key, subscriberMessages, messageId)) {
// no message reference held
removeMessage = true;
if (LOG.isDebugEnabled()) {
@ -198,10 +200,11 @@ public class KahaTopicReferenceStore extends KahaReferenceStore implements Topic
//
// see: https://issues.apache.org/activemq/browse/AMQ-2123
private boolean isUnreferencedBySubscribers(
Map<String, TopicSubContainer> subscriberContainers, MessageId messageId) {
String key, Map<String, TopicSubContainer> subscriberContainers, MessageId messageId) {
boolean isUnreferenced = true;
for (TopicSubContainer container: subscriberContainers.values()) {
if (!container.isEmpty()) {
for (Entry<String, TopicSubContainer> entry : subscriberContainers.entrySet()) {
if (!key.equals(entry.getKey()) && !entry.getValue().isEmpty()) {
TopicSubContainer container = entry.getValue();
for (Iterator i = container.iterator(); i.hasNext();) {
ConsumerMessageRef ref = (ConsumerMessageRef) i.next();
if (messageId.equals(ref.getMessageId())) {

9
activemq-core/src/main/java/org/apache/activemq/transaction/LocalTransaction.java
View File

@ -45,6 +45,11 @@ public class LocalTransaction extends Transaction {
}
public void commit(boolean onePhase) throws XAException, IOException {
if (LOG.isDebugEnabled()) {
LOG.debug("commit: " + xid
+ " syncCount: " + size());
}
// Get ready for commit.
try {
prePrepare();
@ -79,6 +84,10 @@ public class LocalTransaction extends Transaction {
public void rollback() throws XAException, IOException {
if (LOG.isDebugEnabled()) {
LOG.debug("rollback: " + xid
+ " syncCount: " + size());
}
setState(Transaction.FINISHED_STATE);
context.getTransactions().remove(xid);
transactionStore.rollback(getTransactionId());

1
activemq-core/src/test/java/org/apache/activemq/advisory/MasterSlaveTempQueueMemoryTest.java
View File

@ -29,7 +29,6 @@ import org.apache.activemq.ActiveMQSession;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.region.Queue;
import org.apache.activemq.broker.region.RegionBroker;
import org.apache.activemq.command.ActiveMQDestination;
public class MasterSlaveTempQueueMemoryTest extends TempQueueMemoryTest {

6
activemq-core/src/test/java/org/apache/activemq/bugs/AMQ1917Test.java
View File

@ -125,8 +125,10 @@ public class AMQ1917Test extends TestCase {
sender.send(response);
}
} catch (JMSException e) {
errorLatch.countDown();
fail("Unexpected exception:" + e);
if (working) {
errorLatch.countDown();
fail("Unexpected exception:" + e);
}
}
}
}

144
activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2149Test.java
View File

@ -39,6 +39,8 @@ import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.region.Destination;
import org.apache.activemq.broker.region.DestinationStatistics;
import org.apache.activemq.broker.region.RegionBroker;
import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.broker.util.LoggingBrokerPlugin;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.store.amq.AMQPersistenceAdapterFactory;
@ -55,20 +57,26 @@ public class AMQ2149Test extends TestCase {
private static final Log LOG = LogFactory.getLog(AMQ2149Test.class);
private static final long BROKER_STOP_PERIOD = 20 * 1000;
private static final String BROKER_CONNECTOR = "tcp://localhost:61617";
private static final String BROKER_URL = "failover:("+ BROKER_CONNECTOR
private static final String DEFAULT_BROKER_URL = "failover:("+ BROKER_CONNECTOR
+")?maxReconnectDelay=1000&useExponentialBackOff=false";
private final String SEQ_NUM_PROPERTY = "seqNum";
final int MESSAGE_LENGTH_BYTES = 75 * 1024;
final int MAX_TO_SEND = 1500;
final long SLEEP_BETWEEN_SEND_MS = 3;
final int NUM_SENDERS_AND_RECEIVERS = 10;
final Object brokerLock = new Object();
private static final long DEFAULT_BROKER_STOP_PERIOD = 20 * 1000;
private static final long DEFAULT_NUM_TO_SEND = 1500;
long brokerStopPeriod = DEFAULT_BROKER_STOP_PERIOD;
long numtoSend = DEFAULT_NUM_TO_SEND;
String brokerURL = DEFAULT_BROKER_URL;
int numBrokerRestarts = 0;
final static int MAX_BROKER_RESTARTS = 4;
BrokerService broker;
Vector<Throwable> exceptions = new Vector<Throwable>();
@ -100,12 +108,17 @@ public class AMQ2149Test extends TestCase {
public void setUp() throws Exception {
dataDirFile = new File("target/"+ getName());
numtoSend = DEFAULT_NUM_TO_SEND;
brokerStopPeriod = DEFAULT_BROKER_STOP_PERIOD;
brokerURL = DEFAULT_BROKER_URL;
}
public void tearDown() throws Exception {
synchronized(brokerLock) {
broker.stop();
broker.waitUntilStopped();
if (broker!= null) {
broker.stop();
broker.waitUntilStopped();
}
}
exceptions.clear();
}
@ -130,15 +143,18 @@ public class AMQ2149Test extends TestCase {
private final MessageConsumer messageConsumer;
private volatile long nextExpectedSeqNum = 0;
private final boolean transactional;
private String lastId = null;
public Receiver(javax.jms.Destination dest) throws JMSException {
public Receiver(javax.jms.Destination dest, boolean transactional) throws JMSException {
this.dest = dest;
connection = new ActiveMQConnectionFactory(BROKER_URL)
this.transactional = transactional;
connection = new ActiveMQConnectionFactory(brokerURL)
.createConnection();
connection.setClientID(dest.toString());
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
session = connection.createSession(transactional, transactional ? Session.SESSION_TRANSACTED : Session.AUTO_ACKNOWLEDGE);
if (ActiveMQDestination.transform(dest).isTopic()) {
messageConsumer = session.createDurableSubscriber((Topic) dest, dest.toString());
} else {
@ -161,6 +177,11 @@ public class AMQ2149Test extends TestCase {
final long seqNum = message.getLongProperty(SEQ_NUM_PROPERTY);
if ((seqNum % 500) == 0) {
LOG.info(dest + " received " + seqNum);
if (transactional) {
LOG.info("committing..");
session.commit();
}
}
if (seqNum != nextExpectedSeqNum) {
LOG.warn(dest + " received " + seqNum
@ -196,7 +217,7 @@ public class AMQ2149Test extends TestCase {
public Sender(javax.jms.Destination dest) throws JMSException {
this.dest = dest;
connection = new ActiveMQConnectionFactory(BROKER_URL)
connection = new ActiveMQConnectionFactory(brokerURL)
.createConnection();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
messageProducer = session.createProducer(dest);
@ -206,7 +227,7 @@ public class AMQ2149Test extends TestCase {
public void run() {
final String longString = buildLongString();
while (nextSequenceNumber < MAX_TO_SEND) {
while (nextSequenceNumber < numtoSend) {
try {
final Message message = session
.createTextMessage(longString);
@ -214,6 +235,11 @@ public class AMQ2149Test extends TestCase {
nextSequenceNumber);
++nextSequenceNumber;
messageProducer.send(message);
if ((nextSequenceNumber % 500) == 0) {
LOG.info(dest + " sent " + nextSequenceNumber);
}
} catch (Exception e) {
LOG.error(dest + " send error", e);
exceptions.add(e);
@ -296,8 +322,7 @@ public class AMQ2149Test extends TestCase {
verifyStats(true);
}
public void testTopicOrderWithRestart() throws Exception {
createBroker(new Configurer() {
public void configure(BrokerService broker) throws Exception {
@ -317,6 +342,54 @@ public class AMQ2149Test extends TestCase {
verifyStats(true);
}
public void testQueueTransactionalOrderWithRestart() throws Exception {
doTestTransactionalOrderWithRestart(ActiveMQDestination.QUEUE_TYPE);
}
public void testTopicTransactionalOrderWithRestart() throws Exception {
doTestTransactionalOrderWithRestart(ActiveMQDestination.TOPIC_TYPE);
}
public void doTestTransactionalOrderWithRestart(byte destinationType) throws Exception {
// with transactions there may be lots of re deliveries, in the case
// or a commit every 500 messages there could be up to 500 re deliveries
// In order to ensure these are acked and don't block new message receipt,
// the prefetch should be less than double the commit window.
// In addition there needs to be sufficient memory to available to dispatch
// transaction size + redeliveries - so 2*transaction size
brokerURL = DEFAULT_BROKER_URL + "&jms.prefetchPolicy.all=240";
numtoSend = 15000;
brokerStopPeriod = 30 * 1000;
final PolicyMap policyMap = new PolicyMap();
PolicyEntry policy = new PolicyEntry();
policy.setMaxPageSize(500);
policyMap.setDefaultEntry(policy);
createBroker(new Configurer() {
public void configure(BrokerService broker) throws Exception {
broker.deleteAllMessages();
broker.setDestinationPolicy(policyMap);
}
});
final Timer timer = new Timer();
schedualRestartTask(timer, new Configurer() {
public void configure(BrokerService broker) throws Exception {
broker.setDestinationPolicy(policyMap);
}
});
try {
verifyOrderedMessageReceipt(destinationType, 1, true);
} finally {
timer.cancel();
}
verifyStats(true);
}
// no need to run this unless there are issues with the other restart tests
@ -356,8 +429,11 @@ public class AMQ2149Test extends TestCase {
for (Destination dest : regionBroker.getQueueRegion().getDestinationMap().values()) {
DestinationStatistics stats = dest.getDestinationStatistics();
if (brokerRestarts) {
assertTrue("qneue/dequeue match for: " + dest.getName(),
stats.getEnqueues().getCount() <= stats.getDequeues().getCount());
// all bets are off w.r.t stats as there may be duplicate sends and duplicate
// dispatches, all of which will be suppressed - either by the reference store
// not allowing duplicate references or consumers acking duplicates
LOG.info("with restart: not asserting qneue/dequeue stat match for: " + dest.getName()
+ " " + stats.getEnqueues().getCount() + " <= " +stats.getDequeues().getCount());
} else {
assertEquals("qneue/dequeue match for: " + dest.getName(),
stats.getEnqueues().getCount(), stats.getDequeues().getCount());
@ -386,29 +462,37 @@ public class AMQ2149Test extends TestCase {
exceptions.add(e);
}
}
// do it again
try {
timer.schedule(new RestartTask(), BROKER_STOP_PERIOD);
} catch (IllegalStateException ignore_alreadyCancelled) {
if (++numBrokerRestarts < MAX_BROKER_RESTARTS) {
// do it again
try {
timer.schedule(new RestartTask(), brokerStopPeriod);
} catch (IllegalStateException ignore_alreadyCancelled) {
}
} else {
LOG.info("no longer stopping broker on reaching Max restarts: " + MAX_BROKER_RESTARTS);
}
}
}
timer.schedule(new RestartTask(), BROKER_STOP_PERIOD);
}
private void verifyOrderedMessageReceipt() throws Exception {
verifyOrderedMessageReceipt(ActiveMQDestination.QUEUE_TYPE);
timer.schedule(new RestartTask(), brokerStopPeriod);
}
private void verifyOrderedMessageReceipt(byte destinationType) throws Exception {
verifyOrderedMessageReceipt(destinationType, NUM_SENDERS_AND_RECEIVERS, false);
}
private void verifyOrderedMessageReceipt() throws Exception {
verifyOrderedMessageReceipt(ActiveMQDestination.QUEUE_TYPE, NUM_SENDERS_AND_RECEIVERS, false);
}
private void verifyOrderedMessageReceipt(byte destinationType, int concurrentPairs, boolean transactional) throws Exception {
Vector<Thread> threads = new Vector<Thread>();
Vector<Receiver> receivers = new Vector<Receiver>();
for (int i = 0; i < NUM_SENDERS_AND_RECEIVERS; ++i) {
for (int i = 0; i < concurrentPairs; ++i) {
final javax.jms.Destination destination =
ActiveMQDestination.createDestination("test.dest." + i, destinationType);
receivers.add(new Receiver(destination));
receivers.add(new Receiver(destination, transactional));
Thread thread = new Thread(new Sender(destination));
thread.start();
threads.add(thread);
@ -426,7 +510,7 @@ public class AMQ2149Test extends TestCase {
while(!receivers.isEmpty() && System.currentTimeMillis() < expiry) {
Receiver receiver = receivers.firstElement();
if (receiver.getNextExpectedSeqNo() >= MAX_TO_SEND || !exceptions.isEmpty()) {
if (receiver.getNextExpectedSeqNo() >= numtoSend || !exceptions.isEmpty()) {
receiver.close();
receivers.remove(receiver);
}