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

This closes #26 error prone errors and fixes on activemq5-tests

This commit is contained in:
Clebert Suconic 2015-06-12 10:38:24 -04:00
parent 145fc7f82c 4dd54080a1
commit 29d59e970b
52 changed files with 234 additions and 235 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tests/activemq5-unit-tests/pom.xml
View File

@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.activemq.tests</groupId>
<artifactId>artemis-tests-pom</artifactId>
<version>1.0.0-SNAPSHOT</version>
<version>1.0.1-SNAPSHOT</version>
</parent>
<artifactId>activemq5-unit-tests</artifactId>

2
tests/activemq5-unit-tests/src/main/java/org/apache/activemq/broker/artemiswrapper/ArtemisBrokerWrapper.java
View File

@ -38,7 +38,7 @@ import org.apache.activemq.broker.BrokerService;
public class ArtemisBrokerWrapper extends ArtemisBrokerBase
{
protected Map<String, SimpleString> testQueues = new HashMap<String, SimpleString>();
protected final Map<String, SimpleString> testQueues = new HashMap<String, SimpleString>();
public ArtemisBrokerWrapper(BrokerService brokerService)
{

6
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsCreateConsumerInOnMessageTest.java
View File

@ -36,7 +36,7 @@ public class JmsCreateConsumerInOnMessageTest extends TestSupport implements Mes
private MessageConsumer testConsumer;
private MessageProducer producer;
private Topic topic;
private Object lock = new Object();
private final Object lock = new Object();
/*
* @see junit.framework.TestCase#setUp()
@ -71,8 +71,8 @@ public class JmsCreateConsumerInOnMessageTest extends TestSupport implements Mes
public void testCreateConsumer() throws Exception {
Message msg = super.createMessage();
producer.send(msg);
if (testConsumer == null) {
synchronized (lock) {
synchronized (lock) {
while(testConsumer == null) {
lock.wait(3000);
}
}

4
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsMultipleClientsTestSupport.java
View File

@ -83,8 +83,6 @@ public class JmsMultipleClientsTestSupport {
protected List<Connection> connections = Collections.synchronizedList(new ArrayList<Connection>());
protected MessageIdList allMessagesList = new MessageIdList();
private AtomicInteger producerLock;
protected void startProducers(Destination dest, int msgCount) throws Exception {
startProducers(createConnectionFactory(), dest, msgCount);
}
@ -92,7 +90,7 @@ public class JmsMultipleClientsTestSupport {
protected void startProducers(final ConnectionFactory factory, final Destination dest, final int msgCount) throws Exception {
// Use concurrent send
if (useConcurrentSend) {
producerLock = new AtomicInteger(producerCount);
final AtomicInteger producerLock = new AtomicInteger(producerCount);
for (int i = 0; i < producerCount; i++) {
Thread t = new Thread(new Runnable() {

8
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/LargeMessageTestSupport.java
View File

@ -61,7 +61,7 @@ public class LargeMessageTestSupport extends ClientTestSupport implements Messag
protected int deliveryMode = DeliveryMode.PERSISTENT;
protected IdGenerator idGen = new IdGenerator();
protected boolean validMessageConsumption = true;
protected AtomicInteger messageCount = new AtomicInteger(0);
protected final AtomicInteger messageCount = new AtomicInteger(0);
protected int prefetchValue = 10000000;
@ -182,9 +182,9 @@ public class LargeMessageTestSupport extends ClientTestSupport implements Messag
producer.send(msg);
}
long now = System.currentTimeMillis();
while (now + 60000 > System.currentTimeMillis() && messageCount.get() < MESSAGE_COUNT) {
LOG.info("message count = " + messageCount);
synchronized (messageCount) {
synchronized (messageCount) {
while (now + 60000 > System.currentTimeMillis() && messageCount.get() < MESSAGE_COUNT) {
LOG.info("message count = " + messageCount);
messageCount.wait(1000);
}
}

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/OnePrefetchAsyncConsumerTest.java
View File

@ -154,7 +154,7 @@ public class OnePrefetchAsyncConsumerTest extends EmbeddedBrokerTestSupport {
}
private class TestServerSession implements ServerSession {
TestServerSessionPool pool;
final TestServerSessionPool pool;
Session session;
public TestServerSession(TestServerSessionPool pool) throws JMSException {

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/broker/ft/DbRestartJDBCQueueMasterSlaveLeaseTest.java
View File

@ -43,7 +43,7 @@ public class DbRestartJDBCQueueMasterSlaveLeaseTest extends DbRestartJDBCQueueMa
LeaseLockerIOExceptionHandler ioExceptionHandler = new LeaseLockerIOExceptionHandler();
ioExceptionHandler.setIgnoreSQLExceptions(false);
ioExceptionHandler.setStopStartConnectors(false);
ioExceptionHandler.setResumeCheckSleepPeriod(500l);
ioExceptionHandler.setResumeCheckSleepPeriod(500L);
brokerService.setIoExceptionHandler(ioExceptionHandler);
}

4
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/broker/policy/AbortSlowConsumer1Test.java
View File

@ -82,7 +82,7 @@ public class AbortSlowConsumer1Test extends AbortSlowConsumerBase {
consumer.close();
TimeUnit.SECONDS.sleep(5);
assertTrue("no exceptions : " + exceptions.toArray(), exceptions.isEmpty());
assertTrue("no exceptions : " + exceptions, exceptions.isEmpty());
}
@Test(timeout = 60 * 1000)
@ -99,6 +99,6 @@ public class AbortSlowConsumer1Test extends AbortSlowConsumerBase {
conn.close();
TimeUnit.SECONDS.sleep(5);
assertTrue("no exceptions : " + exceptions.toArray(), exceptions.isEmpty());
assertTrue("no exceptions : " + exceptions, exceptions.isEmpty());
}
}

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/broker/region/QueuePurgeTest.java
View File

@ -67,7 +67,7 @@ public class QueuePurgeTest extends CombinationTestSupport {
broker.setDataDirectoryFile(testDataDir);
broker.setUseJmx(true);
broker.setDeleteAllMessagesOnStartup(true);
broker.getSystemUsage().getMemoryUsage().setLimit(1024l*1024*64);
broker.getSystemUsage().getMemoryUsage().setLimit(1024L*1024*64);
KahaDBPersistenceAdapter persistenceAdapter = new KahaDBPersistenceAdapter();
persistenceAdapter.setDirectory(new File(testDataDir, "kahadb"));
broker.setPersistenceAdapter(persistenceAdapter);

4
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/broker/region/QueueResendDuringShutdownTest.java
View File

@ -54,7 +54,7 @@ public class QueueResendDuringShutdownTest {
private Connection producerConnection;
private Queue queue;
private Object messageReceiveSync = new Object();
private final Object messageReceiveSync = new Object();
private int receiveCount;
@Before
@ -239,7 +239,7 @@ public class QueueResendDuringShutdownTest {
protected void waitForMessage (long delayMs) {
try {
synchronized ( this.messageReceiveSync ) {
if ( this.receiveCount == 0 ) {
while ( this.receiveCount == 0 ) {
this.messageReceiveSync.wait(delayMs);
}
}

46
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/StoreQueueCursorOrderTest.java
View File

@ -106,7 +106,7 @@ public class StoreQueueCursorOrderTest {
@Override
public void run() {
}
}, 2l) {};
}, 2L) {};
msg.getMessageId().setFutureOrSequenceLong(future);
underTest.addMessageLast(msg);
@ -116,12 +116,12 @@ public class StoreQueueCursorOrderTest {
msg = getMessage(1);
messages[0] = msg;
msg.setMemoryUsage(systemUsage.getMemoryUsage());
msg.getMessageId().setFutureOrSequenceLong(1l);
msg.getMessageId().setFutureOrSequenceLong(1L);
underTest.addMessageLast(msg);
assertTrue("cache is disabled as limit reached", !underTest.isCacheEnabled());
assertEquals("setBatch unset", 0l, queueMessageStore.batch.get());
assertEquals("setBatch unset", 0L, queueMessageStore.batch.get());
int dequeueCount = 0;
@ -171,9 +171,9 @@ public class StoreQueueCursorOrderTest {
FutureTask<Long> future = new FutureTask<Long>(new Runnable() {
@Override
public void run() {
msgRef.getMessageId().setFutureOrSequenceLong(1l);
msgRef.getMessageId().setFutureOrSequenceLong(1L);
}
}, 1l) {};
}, 1L) {};
msg.getMessageId().setFutureOrSequenceLong(future);
Executors.newSingleThreadExecutor().submit(future);
underTest.addMessageLast(msg);
@ -184,12 +184,12 @@ public class StoreQueueCursorOrderTest {
msg = getMessage(1);
messages[0] = msg;
msg.setMemoryUsage(systemUsage.getMemoryUsage());
msg.getMessageId().setFutureOrSequenceLong(1l);
msg.getMessageId().setFutureOrSequenceLong(1L);
underTest.addMessageLast(msg);
assertTrue("cache is disabled as limit reached", !underTest.isCacheEnabled());
assertEquals("setBatch unset", 0l, queueMessageStore.batch.get());
assertEquals("setBatch unset", 0L, queueMessageStore.batch.get());
int dequeueCount = 0;
@ -239,9 +239,9 @@ public class StoreQueueCursorOrderTest {
FutureTask<Long> future = new FutureTask<Long>(new Runnable() {
@Override
public void run() {
msgRef.getMessageId().setFutureOrSequenceLong(0l);
msgRef.getMessageId().setFutureOrSequenceLong(0L);
}
}, 0l) {};
}, 0L) {};
msg.getMessageId().setFutureOrSequenceLong(future);
Executors.newSingleThreadExecutor().submit(future);
underTest.addMessageLast(msg);
@ -257,16 +257,16 @@ public class StoreQueueCursorOrderTest {
FutureTask<Long> future2 = new FutureTask<Long>(new Runnable() {
@Override
public void run() {
msgRe2f.getMessageId().setFutureOrSequenceLong(1l);
msgRe2f.getMessageId().setFutureOrSequenceLong(1L);
}
}, 1l) {};
}, 1L) {};
msg.getMessageId().setFutureOrSequenceLong(future2);
Executors.newSingleThreadExecutor().submit(future2);
underTest.addMessageLast(msg);
assertTrue("cache is disabled as limit reached", !underTest.isCacheEnabled());
assertEquals("setBatch set", 1l, queueMessageStore.batch.get());
assertEquals("setBatch set", 1L, queueMessageStore.batch.get());
int dequeueCount = 0;
@ -316,9 +316,9 @@ public class StoreQueueCursorOrderTest {
FutureTask<Long> future0 = new FutureTask<Long>(new Runnable() {
@Override
public void run() {
msgRef.getMessageId().setFutureOrSequenceLong(0l);
msgRef.getMessageId().setFutureOrSequenceLong(0L);
}
}, 0l) {};
}, 0L) {};
msg.getMessageId().setFutureOrSequenceLong(future0);
underTest.addMessageLast(msg);
Executors.newSingleThreadExecutor().submit(future0);
@ -332,9 +332,9 @@ public class StoreQueueCursorOrderTest {
FutureTask<Long> future1 = new FutureTask<Long>(new Runnable() {
@Override
public void run() {
msgRef1.getMessageId().setFutureOrSequenceLong(3l);
msgRef1.getMessageId().setFutureOrSequenceLong(3L);
}
}, 3l) {};
}, 3L) {};
msg.getMessageId().setFutureOrSequenceLong(future1);
underTest.addMessageLast(msg);
@ -342,7 +342,7 @@ public class StoreQueueCursorOrderTest {
msg = getMessage(2);
messages[1] = msg;
msg.setMemoryUsage(systemUsage.getMemoryUsage());
msg.getMessageId().setFutureOrSequenceLong(1l);
msg.getMessageId().setFutureOrSequenceLong(1L);
underTest.addMessageLast(msg);
assertTrue("cache enabled", underTest.isUseCache() && underTest.isCacheEnabled());
@ -354,12 +354,12 @@ public class StoreQueueCursorOrderTest {
msg = getMessage(3);
messages[2] = msg;
msg.setMemoryUsage(systemUsage.getMemoryUsage());
msg.getMessageId().setFutureOrSequenceLong(3l);
msg.getMessageId().setFutureOrSequenceLong(3L);
underTest.addMessageLast(msg);
assertTrue("cache is disabled as limit reached", !underTest.isCacheEnabled());
assertEquals("setBatch set", 2l, queueMessageStore.batch.get());
assertEquals("setBatch set", 2L, queueMessageStore.batch.get());
int dequeueCount = 0;
@ -405,13 +405,13 @@ public class StoreQueueCursorOrderTest {
ActiveMQTextMessage msg = getMessage(0);
messages[0] = msg;
msg.setMemoryUsage(systemUsage.getMemoryUsage());
msg.getMessageId().setFutureOrSequenceLong(0l);
msg.getMessageId().setFutureOrSequenceLong(0L);
underTest.addMessageLast(msg);
msg = getMessage(1);
messages[1] = msg;
msg.setMemoryUsage(systemUsage.getMemoryUsage());
msg.getMessageId().setFutureOrSequenceLong(1l);
msg.getMessageId().setFutureOrSequenceLong(1L);
underTest.addMessageLast(msg);
assertTrue("cache enabled", underTest.isUseCache() && underTest.isCacheEnabled());
@ -419,12 +419,12 @@ public class StoreQueueCursorOrderTest {
msg = getMessage(2);
messages[2] = msg;
msg.setMemoryUsage(systemUsage.getMemoryUsage());
msg.getMessageId().setFutureOrSequenceLong(2l);
msg.getMessageId().setFutureOrSequenceLong(2L);
underTest.addMessageLast(msg);
assertTrue("cache is disabled as limit reached", !underTest.isCacheEnabled());
assertEquals("setBatch set", 2l, queueMessageStore.batch.get());
assertEquals("setBatch set", 2L, queueMessageStore.batch.get());
int dequeueCount = 0;

8
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/broker/scheduler/JmsSchedulerTest.java
View File

@ -260,7 +260,7 @@ public class JmsSchedulerTest extends JobSchedulerTestSupport {
// wait for the producer to block, which should happen immediately, and also wait long
// enough for the delay to elapse. We should see no deliveries as the send should block
// on the first message.
Thread.sleep(10000l);
Thread.sleep(10000L);
assertEquals(100, latch.getCount());
@ -268,7 +268,7 @@ public class JmsSchedulerTest extends JobSchedulerTestSupport {
broker.getSystemUsage().getJobSchedulerUsage().setLimit(1024 * 1024 * 33);
// Wait long enough that the messages are enqueued and the delivery delay has elapsed.
Thread.sleep(10000l);
Thread.sleep(10000L);
// Make sure we sent all the messages we expected to send
Wait.waitFor(new Wait.Condition() {
@ -276,12 +276,12 @@ public class JmsSchedulerTest extends JobSchedulerTestSupport {
public boolean isSatisified() throws Exception {
return producer.getSentCount() == producer.getMessageCount();
}
}, 20000l);
}, 20000L);
assertEquals("Producer didn't send all messages", producer.getMessageCount(), producer.getSentCount());
// Make sure we got all the messages we expected to get
latch.await(20000l, TimeUnit.MILLISECONDS);
latch.await(20000L, TimeUnit.MILLISECONDS);
assertEquals("Consumer did not receive all messages.", 0, latch.getCount());
}

114
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ2149Test.java
View File

@ -30,6 +30,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
import javax.jms.*;
@ -65,22 +66,22 @@ public class AMQ2149Test {
private static final String BROKER_CONNECTOR = "tcp://localhost:61617";
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 long SLEEP_BETWEEN_SEND_MS = 25;
final int NUM_SENDERS_AND_RECEIVERS = 10;
final Object brokerLock = new Object();
private static final long DEFAULT_BROKER_STOP_PERIOD = 10 * 1000;
private static final long DEFAULT_NUM_TO_SEND = 1400;
long brokerStopPeriod = DEFAULT_BROKER_STOP_PERIOD;
long numtoSend = DEFAULT_NUM_TO_SEND;
long sleepBetweenSend = SLEEP_BETWEEN_SEND_MS;
String brokerURL = DEFAULT_BROKER_URL;
int numBrokerRestarts = 0;
final static int MAX_BROKER_RESTARTS = 4;
BrokerService broker;
@ -88,15 +89,15 @@ public class AMQ2149Test {
protected File dataDirFile;
final LoggingBrokerPlugin[] plugins = new LoggingBrokerPlugin[]{new LoggingBrokerPlugin()};
public void createBroker(Configurer configurer) throws Exception {
broker = new BrokerService();
configurePersistenceAdapter(broker);
broker.getSystemUsage().getMemoryUsage().setLimit(MESSAGE_LENGTH_BYTES * 200 * NUM_SENDERS_AND_RECEIVERS);
broker.addConnector(BROKER_CONNECTOR);
broker.addConnector(BROKER_CONNECTOR);
broker.setBrokerName(testName.getMethodName());
broker.setDataDirectoryFile(dataDirFile);
if (configurer != null) {
@ -104,7 +105,7 @@ public class AMQ2149Test {
}
broker.start();
}
protected void configurePersistenceAdapter(BrokerService brokerService) throws Exception {
}
@ -135,7 +136,7 @@ public class AMQ2149Test {
executor.shutdownNow();
exceptions.clear();
}
private String buildLongString() {
final StringBuilder stringBuilder = new StringBuilder(
MESSAGE_LENGTH_BYTES);
@ -156,8 +157,8 @@ public class AMQ2149Test {
private final MessageConsumer messageConsumer;
private volatile long nextExpectedSeqNum = 0;
private AtomicLong nextExpectedSeqNum = new AtomicLong();
private final boolean transactional;
private String lastId = null;
@ -182,11 +183,11 @@ public class AMQ2149Test {
public void close() throws JMSException {
connection.close();
}
public long getNextExpectedSeqNo() {
return nextExpectedSeqNum;
return nextExpectedSeqNum.get();
}
final int TRANSACITON_BATCH = 500;
boolean resumeOnNextOrPreviousIsOk = false;
public void onMessage(Message message) {
@ -194,7 +195,7 @@ public class AMQ2149Test {
final long seqNum = message.getLongProperty(SEQ_NUM_PROPERTY);
if ((seqNum % TRANSACITON_BATCH) == 0) {
LOG.info(dest + " received " + seqNum);
if (transactional) {
LOG.info("committing..");
session.commit();
@ -202,25 +203,26 @@ public class AMQ2149Test {
}
if (resumeOnNextOrPreviousIsOk) {
// after an indoubt commit we need to accept what we get (within reason)
if (seqNum != nextExpectedSeqNum) {
if (seqNum == nextExpectedSeqNum - (TRANSACITON_BATCH -1)) {
nextExpectedSeqNum -= (TRANSACITON_BATCH -1);
if (seqNum != nextExpectedSeqNum.get()) {
final long l = nextExpectedSeqNum.get();
if (seqNum == l - (TRANSACITON_BATCH -1)) {
nextExpectedSeqNum.compareAndSet(l, l - (TRANSACITON_BATCH -1) );
LOG.info("In doubt commit failed, getting replay at:" + nextExpectedSeqNum);
}
}
resumeOnNextOrPreviousIsOk = false;
}
if (seqNum != nextExpectedSeqNum) {
if (seqNum != nextExpectedSeqNum.get()) {
LOG.warn(dest + " received " + seqNum
+ " in msg: " + message.getJMSMessageID()
+ " expected "
+ nextExpectedSeqNum
+ ", lastId: " + lastId
+ ", lastId: " + lastId
+ ", message:" + message);
fail(dest + " received " + seqNum + " expected "
+ nextExpectedSeqNum);
}
++nextExpectedSeqNum;
nextExpectedSeqNum.incrementAndGet();
lastId = message.getJMSMessageID();
} catch (TransactionRolledBackException expectedSometimesOnFailoverRecovery) {
LOG.info("got rollback: " + expectedSometimesOnFailoverRecovery);
@ -228,12 +230,12 @@ public class AMQ2149Test {
// in doubt - either commit command or reply missing
// don't know if we will get a replay
resumeOnNextOrPreviousIsOk = true;
nextExpectedSeqNum++;
nextExpectedSeqNum.incrementAndGet();
LOG.info("in doubt transaction completion: ok to get next or previous batch. next:" + nextExpectedSeqNum);
} else {
resumeOnNextOrPreviousIsOk = false;
// batch will be replayed
nextExpectedSeqNum -= (TRANSACITON_BATCH -1);
nextExpectedSeqNum.addAndGet(-(TRANSACITON_BATCH - 1));
}
} catch (Throwable e) {
@ -255,6 +257,7 @@ public class AMQ2149Test {
private final MessageProducer messageProducer;
private volatile long nextSequenceNumber = 0;
private final Object guard = new Object();
public Sender(javax.jms.Destination dest) throws JMSException {
this.dest = dest;
@ -269,15 +272,24 @@ public class AMQ2149Test {
public void run() {
final String longString = buildLongString();
long nextSequenceNumber = this.nextSequenceNumber;
while (nextSequenceNumber < numtoSend) {
try {
final Message message = session
.createTextMessage(longString);
message.setLongProperty(SEQ_NUM_PROPERTY,
nextSequenceNumber);
++nextSequenceNumber;
messageProducer.send(message);
synchronized (guard)
{
if (nextSequenceNumber == this.nextSequenceNumber)
{
this.nextSequenceNumber = nextSequenceNumber + 1;
messageProducer.send(message);
} else {
continue;
}
}
if ((nextSequenceNumber % 500) == 0) {
LOG.info(dest + " sent " + nextSequenceNumber);
}
@ -353,13 +365,13 @@ public class AMQ2149Test {
// no need to run this unless there are some issues with the others
public void vanilaVerify_testOrder() throws Exception {
createBroker(new Configurer() {
public void configure(BrokerService broker) throws Exception {
broker.deleteAllMessages();
broker.deleteAllMessages();
}
});
verifyOrderedMessageReceipt();
verifyStats(false);
}
@ -368,22 +380,22 @@ public class AMQ2149Test {
public void testOrderWithRestart() throws Exception {
createBroker(new Configurer() {
public void configure(BrokerService broker) throws Exception {
broker.deleteAllMessages();
broker.deleteAllMessages();
}
});
final Timer timer = new Timer();
schedualRestartTask(timer, new Configurer() {
public void configure(BrokerService broker) throws Exception {
public void configure(BrokerService broker) throws Exception {
}
});
try {
verifyOrderedMessageReceipt();
} finally {
timer.cancel();
}
verifyStats(true);
}
@ -394,16 +406,16 @@ public class AMQ2149Test {
broker.deleteAllMessages();
}
});
final Timer timer = new Timer();
schedualRestartTask(timer, null);
try {
verifyOrderedMessageReceipt(ActiveMQDestination.TOPIC_TYPE);
} finally {
timer.cancel();
}
verifyStats(true);
}
@ -416,33 +428,33 @@ public class AMQ2149Test {
public void testTopicTransactionalOrderWithRestart() throws Exception {
doTestTransactionalOrderWithRestart(ActiveMQDestination.TOPIC_TYPE);
}
public void doTestTransactionalOrderWithRestart(byte destinationType) throws Exception {
numtoSend = 10000;
sleepBetweenSend = 3;
brokerStopPeriod = 10 * 1000;
createBroker(new Configurer() {
public void configure(BrokerService broker) throws Exception {
broker.deleteAllMessages();
}
});
final Timer timer = new Timer();
schedualRestartTask(timer, null);
try {
verifyOrderedMessageReceipt(destinationType, 1, true);
} finally {
timer.cancel();
}
verifyStats(true);
}
private void verifyStats(boolean brokerRestarts) throws Exception {
RegionBroker regionBroker = (RegionBroker) broker.getRegionBroker();
for (Destination dest : regionBroker.getQueueRegion().getDestinationMap().values()) {
DestinationStatistics stats = dest.getDestinationStatistics();
if (brokerRestarts) {
@ -453,7 +465,7 @@ public class AMQ2149Test {
+ " " + stats.getEnqueues().getCount() + " <= " +stats.getDequeues().getCount());
} else {
assertEquals("qneue/dequeue match for: " + dest.getName(),
stats.getEnqueues().getCount(), stats.getDequeues().getCount());
stats.getEnqueues().getCount(), stats.getDequeues().getCount());
}
}
}
@ -496,20 +508,20 @@ public class AMQ2149Test {
}
return task;
}
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 < concurrentPairs; ++i) {
final javax.jms.Destination destination =
ActiveMQDestination.createDestination("test.dest." + i, destinationType);
@ -518,7 +530,7 @@ public class AMQ2149Test {
thread.start();
threads.add(thread);
}
final long expiry = System.currentTimeMillis() + 1000 * 60 * 4;
while(!threads.isEmpty() && exceptions.isEmpty() && System.currentTimeMillis() < expiry) {
Thread sendThread = threads.firstElement();
@ -563,7 +575,7 @@ public class AMQ2149Test {
}
class TeardownTask implements Callable<Boolean> {
private Object brokerLock;
private final Object brokerLock;
private BrokerService broker;
public TeardownTask(Object brokerLock, BrokerService broker) {

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ2314Test.java
View File

@ -157,7 +157,7 @@ public class AMQ2314Test extends CombinationTestSupport {
broker.setUseJmx(true);
broker.setAdvisorySupport(false);
broker.setDeleteAllMessagesOnStartup(true);
broker.getSystemUsage().getMemoryUsage().setLimit(1024l*1024*64);
broker.getSystemUsage().getMemoryUsage().setLimit(1024L*1024*64);
broker.addConnector("tcp://localhost:0").setName("Default");
broker.start();

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ3779Test.java
View File

@ -47,7 +47,7 @@ public class AMQ3779Test extends AutoFailTestSupport {
}
}
};
logger.getRootLogger().addAppender(appender);
Logger.getRootLogger().addAppender(appender);
try {

8
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4485LowLimitTest.java
View File

@ -101,14 +101,14 @@ public class AMQ4485LowLimitTest extends JmsMultipleBrokersTestSupport {
addNetworkConnector(broker);
}
broker.setSchedulePeriodForDestinationPurge(0);
broker.getSystemUsage().getMemoryUsage().setLimit(256 * 1024 * 1024l);
broker.getSystemUsage().getMemoryUsage().setLimit(256 * 1024 * 1024L);
PolicyMap policyMap = new PolicyMap();
PolicyEntry policyEntry = new PolicyEntry();
policyEntry.setExpireMessagesPeriod(0);
policyEntry.setQueuePrefetch(1000);
policyEntry.setMemoryLimit(2 * 1024 * 1024l);
policyEntry.setMemoryLimit(2 * 1024 * 1024L);
policyEntry.setProducerFlowControl(false);
policyEntry.setEnableAudit(true);
policyEntry.setUseCache(true);
@ -117,7 +117,7 @@ public class AMQ4485LowLimitTest extends JmsMultipleBrokersTestSupport {
PolicyEntry inPolicyEntry = new PolicyEntry();
inPolicyEntry.setExpireMessagesPeriod(0);
inPolicyEntry.setQueuePrefetch(1000);
inPolicyEntry.setMemoryLimit(5 * 1024 * 1024l);
inPolicyEntry.setMemoryLimit(5 * 1024 * 1024L);
inPolicyEntry.setProducerFlowControl(true);
inPolicyEntry.setEnableAudit(true);
inPolicyEntry.setUseCache(true);
@ -252,7 +252,7 @@ public class AMQ4485LowLimitTest extends JmsMultipleBrokersTestSupport {
}
return true;
}
}, 1000 * 60 * 1000l, 20*1000));
}, 1000 * 60 * 1000L, 20*1000));
assertTrue("No exceptions:" + exceptions, exceptions.isEmpty());

4
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4485NetworkOfXBrokersWithNDestsFanoutTransactionTest.java
View File

@ -98,7 +98,7 @@ public class AMQ4485NetworkOfXBrokersWithNDestsFanoutTransactionTest extends Jms
PolicyEntry policyEntry = new PolicyEntry();
policyEntry.setExpireMessagesPeriod(0);
policyEntry.setQueuePrefetch(1000);
policyEntry.setMemoryLimit(1024 * 1024l);
policyEntry.setMemoryLimit(1024 * 1024L);
policyEntry.setOptimizedDispatch(false);
policyEntry.setProducerFlowControl(false);
policyEntry.setEnableAudit(true);
@ -171,7 +171,7 @@ public class AMQ4485NetworkOfXBrokersWithNDestsFanoutTransactionTest extends Jms
}
return true;
}
}, 1000 * 60 * 1000l));
}, 1000 * 60 * 1000L));
assertTrue("No exceptions:" + exceptions, exceptions.isEmpty());

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4607Test.java
View File

@ -49,7 +49,7 @@ public class AMQ4607Test extends JmsMultipleBrokersTestSupport implements Uncaug
public boolean duplex = true;
protected Map<String, MessageConsumer> consumerMap;
Map<Thread, Throwable> unhandeledExceptions = new HashMap<Thread, Throwable>();
final Map<Thread, Throwable> unhandeledExceptions = new HashMap<Thread, Throwable>();
private void assertNoUnhandeledExceptions() {
for( Entry<Thread, Throwable> e: unhandeledExceptions.entrySet()) {

4
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4636Test.java
View File

@ -101,9 +101,9 @@ public class AMQ4636Test {
JDBCPersistenceAdapter jdbc = new TestJDBCPersistenceAdapter();
jdbc.setDataSource(embeddedDataSource);
jdbc.setLockKeepAlivePeriod(1000l);
jdbc.setLockKeepAlivePeriod(1000L);
LeaseDatabaseLocker leaseDatabaseLocker = new LeaseDatabaseLocker();
leaseDatabaseLocker.setLockAcquireSleepInterval(2000l);
leaseDatabaseLocker.setLockAcquireSleepInterval(2000L);
jdbc.setLocker(leaseDatabaseLocker);
broker = new BrokerService();

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ5266SingleDestTest.java
View File

@ -227,7 +227,7 @@ public class AMQ5266SingleDestTest {
}
// verify empty dlq
assertEquals("No pending messages", 0l, ((RegionBroker) brokerService.getRegionBroker()).getDestinationStatistics().getMessages().getCount());
assertEquals("No pending messages", 0L, ((RegionBroker) brokerService.getRegionBroker()).getDestinationStatistics().getMessages().getCount());
}
public class ExportQueuePublisher {

7
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/bugs/CraigsBugTest.java
View File

@ -25,6 +25,9 @@ import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.EmbeddedBrokerTestSupport;
import org.apache.activemq.command.ActiveMQQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
public class CraigsBugTest extends EmbeddedBrokerTestSupport {
private String connectionUri;
@ -49,9 +52,7 @@ public class CraigsBugTest extends EmbeddedBrokerTestSupport {
conn.start();
try {
synchronized (this) {
wait(3000);
}
new CountDownLatch(1).await(3, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
}

4
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/bugs/TrapMessageInJDBCStoreTest.java
View File

@ -82,9 +82,9 @@ public class TrapMessageInJDBCStoreTest extends TestCase {
jdbc.setCleanupPeriod(0);
testTransactionContext = new TestTransactionContext(jdbc);
jdbc.setLockKeepAlivePeriod(1000l);
jdbc.setLockKeepAlivePeriod(1000L);
LeaseDatabaseLocker leaseDatabaseLocker = new LeaseDatabaseLocker();
leaseDatabaseLocker.setLockAcquireSleepInterval(2000l);
leaseDatabaseLocker.setLockAcquireSleepInterval(2000L);
jdbc.setLocker(leaseDatabaseLocker);
broker.setPersistenceAdapter(jdbc);

5
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/bugs/amq1974/TryJmsClient.java
View File

@ -25,6 +25,7 @@ import org.apache.activemq.transport.discovery.simple.SimpleDiscoveryAgent;
import javax.jms.*;
import java.io.File;
import java.net.URISyntaxException;
import java.util.concurrent.CountDownLatch;
public class TryJmsClient
{
@ -59,9 +60,7 @@ public class TryJmsClient
startMessageSend();
synchronized(this) {
this.wait();
}
new CountDownLatch(1).await();
}
private void startUsageMonitor(final BrokerService brokerService) {

5
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/bugs/amq1974/TryJmsManager.java
View File

@ -25,6 +25,7 @@ import org.apache.activemq.transport.discovery.simple.SimpleDiscoveryAgent;
import javax.jms.*;
import java.io.File;
import java.net.URISyntaxException;
import java.util.concurrent.CountDownLatch;
public class TryJmsManager {
@ -59,9 +60,7 @@ public class TryJmsManager {
startMessageConsumer();
synchronized(this) {
this.wait();
}
new CountDownLatch(1).await();
}
private void startUsageMonitor(final BrokerService brokerService) {

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/command/ActiveMQStreamMessageTest.java
View File

@ -979,7 +979,7 @@ public class ActiveMQStreamMessageTest extends TestCase {
message.writeObject(new Byte((byte) 2));
message.writeObject(new Short((short) 2));
message.writeObject(new Integer(2));
message.writeObject(new Long(2l));
message.writeObject(new Long(2L));
message.writeObject(new Float(2.0f));
message.writeObject(new Double(2.0d));
}catch(Exception e) {

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/command/ActiveMQTextMessageTest.java
View File

@ -49,7 +49,7 @@ public class ActiveMQTextMessageTest extends TestCase {
String string = "str";
msg.setText(string);
Message copy = msg.copy();
assertTrue(msg.getText() == ((ActiveMQTextMessage) copy).getText());
assertSame(msg.getText(), ((ActiveMQTextMessage) copy).getText());
}
public void testSetText() {

33
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/spring/ConsumerBean.java
View File

@ -73,14 +73,19 @@ public class ConsumerBean extends Assert implements MessageListener {
long start = System.currentTimeMillis();
try {
if (hasReceivedMessage()) {
synchronized (messages) {
synchronized(messages)
{
try
{
while (hasReceivedMessage())
{
messages.wait(4000);
}
}
} catch (InterruptedException e) {
LOG.info("Caught: " + e);
catch (InterruptedException e)
{
LOG.info("Caught: " + e);
}
}
long end = System.currentTimeMillis() - start;
@ -101,18 +106,18 @@ public class ConsumerBean extends Assert implements MessageListener {
LOG.info("Waiting for (" + maxRemainingMessageCount + ") message(s) to arrive");
long start = System.currentTimeMillis();
long endTime = start + maxWaitTime;
while (maxRemainingMessageCount > 0) {
try {
synchronized (messages) {
synchronized (messages) {
while (maxRemainingMessageCount > 0) {
try {
messages.wait(1000);
if (hasReceivedMessages(messageCount) || System.currentTimeMillis() > endTime) {
break;
}
} catch (InterruptedException e) {
LOG.info("Caught: " + e);
}
if (hasReceivedMessages(messageCount) || System.currentTimeMillis() > endTime) {
break;
}
} catch (InterruptedException e) {
LOG.info("Caught: " + e);
maxRemainingMessageCount = Math.max(0, messageCount - messages.size());
}
maxRemainingMessageCount = Math.max(0, messageCount - messages.size());
}
long end = System.currentTimeMillis() - start;
LOG.info("End of wait for " + end + " millis");

8
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/spring/SpringConsumer.java
View File

@ -43,13 +43,13 @@ public class SpringConsumer extends ConsumerBean implements MessageListener {
try {
ConnectionFactory factory = template.getConnectionFactory();
connection = factory.createConnection();
final Connection c = connection = factory.createConnection();
// we might be a reusable connection in spring
// so lets only set the client ID once if its not set
synchronized (connection) {
if (connection.getClientID() == null) {
connection.setClientID(myId);
synchronized (c) {
if (c.getClientID() == null) {
c.setClientID(myId);
}
}

12
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerTest.java
View File

@ -71,17 +71,17 @@ public class JDBCIOExceptionHandlerTest extends TestCase {
JDBCPersistenceAdapter jdbc = new JDBCPersistenceAdapter();
jdbc.setDataSource(dataSource);
jdbc.setLockKeepAlivePeriod(1000l);
jdbc.setLockKeepAlivePeriod(1000L);
if (leaseLocker) {
LeaseDatabaseLocker leaseDatabaseLocker = new LeaseDatabaseLocker();
leaseDatabaseLocker.setHandleStartException(true);
leaseDatabaseLocker.setLockAcquireSleepInterval(2000l);
leaseDatabaseLocker.setLockAcquireSleepInterval(2000L);
jdbc.setLocker(leaseDatabaseLocker);
}
broker.setPersistenceAdapter(jdbc);
LeaseLockerIOExceptionHandler ioExceptionHandler = new LeaseLockerIOExceptionHandler();
ioExceptionHandler.setResumeCheckSleepPeriod(1000l);
ioExceptionHandler.setResumeCheckSleepPeriod(1000L);
ioExceptionHandler.setStopStartConnectors(startStopConnectors);
broker.setIoExceptionHandler(ioExceptionHandler);
String connectionUri = broker.addConnector(TRANSPORT_URL).getPublishableConnectString();
@ -129,18 +129,18 @@ public class JDBCIOExceptionHandlerTest extends TestCase {
JDBCPersistenceAdapter jdbc = new JDBCPersistenceAdapter();
jdbc.setDataSource(dataSource);
jdbc.setLockKeepAlivePeriod(1000l);
jdbc.setLockKeepAlivePeriod(1000L);
if (lease) {
LeaseDatabaseLocker leaseDatabaseLocker = new LeaseDatabaseLocker();
leaseDatabaseLocker.setHandleStartException(true);
leaseDatabaseLocker.setLockAcquireSleepInterval(2000l);
leaseDatabaseLocker.setLockAcquireSleepInterval(2000L);
jdbc.setLocker(leaseDatabaseLocker);
}
broker.setPersistenceAdapter(jdbc);
LeaseLockerIOExceptionHandler ioExceptionHandler = new LeaseLockerIOExceptionHandler();
ioExceptionHandler.setResumeCheckSleepPeriod(1000l);
ioExceptionHandler.setResumeCheckSleepPeriod(1000L);
ioExceptionHandler.setStopStartConnectors(false);
broker.setIoExceptionHandler(ioExceptionHandler);
slave.set(broker);

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCMessagePriorityTest.java
View File

@ -150,7 +150,7 @@ public class JDBCMessagePriorityTest extends MessagePriorityTest {
for (int priority = 0; priority < maxPriority; priority++) {
producers.add(new ProducerThread(topic, MSG_NUM, priority));
messageCounts[priority] = new AtomicInteger(0);
messageIds[priority] = 1l;
messageIds[priority] = 1L;
}
for (ProducerThread producer : producers) {

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/LeaseDatabaseLockerTest.java
View File

@ -188,7 +188,7 @@ public class LeaseDatabaseLockerTest {
statement = connection.prepareStatement(jdbc.getStatements().getLeaseUpdateStatement());
statement.setString(1, null);
statement.setLong(2, 0l);
statement.setLong(2, 0L);
statement.setString(3, fakeId);
assertEquals("we released " + fakeId, 1, statement.executeUpdate());
LOG.info("released " + fakeId);

6
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBFastEnqueueTest.java
View File

@ -189,7 +189,7 @@ public class KahaDBFastEnqueueTest {
MessageProducer producer = session.createProducer(destination);
Long start = System.currentTimeMillis();
long i = 0l;
long i = 0L;
while ( (i=count.getAndDecrement()) > 0) {
Message message = null;
if (useBytesMessage) {
@ -238,8 +238,8 @@ public class KahaDBFastEnqueueTest {
public void testRollover() throws Exception {
byte flip = 0x1;
for (long i=0; i<Short.MAX_VALUE; i++) {
assertEquals("0 @:" + i, 0, flip ^= 1);
assertEquals("1 @:" + i, 1, flip ^= 1);
assertEquals("0 @:" + i, 0, flip ^= (byte) 1);
assertEquals("1 @:" + i, 1, flip ^= (byte) 1);
}
}
}

10
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/plist/PListTest.java
View File

@ -22,11 +22,7 @@ import static org.junit.Assert.assertTrue;
import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Vector;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@ -46,7 +42,7 @@ public class PListTest {
private PListStoreImpl store;
private PListImpl plist;
final ByteSequence payload = new ByteSequence(new byte[400]);
final String idSeed = new String("Seed" + new byte[1024]);
final String idSeed = new String("Seed" + Arrays.toString(new byte[1024]));
final Vector<Throwable> exceptions = new Vector<Throwable>();
ExecutorService executor;
@ -617,7 +613,7 @@ public class PListTest {
}
}
Map<PList, Object> locks = new HashMap<PList, Object>();
final Map<PList, Object> locks = new HashMap<PList, Object>();
private Object plistLocks(PList plist) {
Object lock = null;

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/streams/JMSInputStreamTest.java
View File

@ -250,7 +250,7 @@ public class JMSInputStreamTest extends JmsTestSupport {
}
out.flush();
synchronized (complete) {
if (!complete.get()) {
while (!complete.get()) {
complete.wait(30000);
}
}

7
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/StubTransport.java
View File

@ -19,6 +19,7 @@ package org.apache.activemq.transport;
import java.io.IOException;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.activemq.util.ServiceStopper;
@ -29,7 +30,7 @@ import org.apache.activemq.util.ServiceStopper;
public class StubTransport extends TransportSupport {
private Queue<Object> queue = new ConcurrentLinkedQueue<Object>();
private volatile int receiveCounter;
private AtomicInteger receiveCounter;
protected void doStop(ServiceStopper stopper) throws Exception {
}
@ -38,7 +39,7 @@ public class StubTransport extends TransportSupport {
}
public void oneway(Object command) throws IOException {
receiveCounter++;
receiveCounter.incrementAndGet();
queue.add(command);
}
@ -51,7 +52,7 @@ public class StubTransport extends TransportSupport {
}
public int getReceiveCounter() {
return receiveCounter;
return receiveCounter.get();
}
}

4
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/TopicClusterTest.java
View File

@ -55,7 +55,7 @@ public class TopicClusterTest extends TestCase implements MessageListener {
protected Destination destination;
protected boolean topic = true;
protected AtomicInteger receivedMessageCount = new AtomicInteger(0);
protected final AtomicInteger receivedMessageCount = new AtomicInteger(0);
protected int deliveryMode = DeliveryMode.NON_PERSISTENT;
protected MessageProducer[] producers;
protected Connection[] connections;
@ -166,7 +166,7 @@ public class TopicClusterTest extends TestCase implements MessageListener {
}
}
synchronized (receivedMessageCount) {
if (receivedMessageCount.get() < expectedReceiveCount()) {
while (receivedMessageCount.get() < expectedReceiveCount()) {
receivedMessageCount.wait(20000);
}
}

27
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/failover/AMQ1925Test.java
View File

@ -21,6 +21,7 @@ import java.net.URI;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.jms.Connection;
@ -73,14 +74,12 @@ public class AMQ1925Test extends TestCase implements ExceptionListener {
// The runnable is likely to interrupt during the session#commit, since
// this takes the longest
final Object starter = new Object();
final CountDownLatch starter = new CountDownLatch(1);
final AtomicBoolean restarted = new AtomicBoolean();
new Thread(new Runnable() {
public void run() {
try {
synchronized (starter) {
starter.wait();
}
starter.await();
// Simulate broker failure & restart
bs.stop();
@ -97,9 +96,6 @@ public class AMQ1925Test extends TestCase implements ExceptionListener {
}
}).start();
synchronized (starter) {
starter.notifyAll();
}
for (int i = 0; i < MESSAGE_COUNT; i++) {
Message message = consumer.receive(500);
assertNotNull("No Message " + i + " found", message);
@ -108,9 +104,7 @@ public class AMQ1925Test extends TestCase implements ExceptionListener {
assertFalse("Timing problem, restarted too soon", restarted
.get());
if (i == 10) {
synchronized (starter) {
starter.notifyAll();
}
starter.countDown();
}
if (i > MESSAGE_COUNT - 100) {
assertTrue("Timing problem, restarted too late", restarted
@ -143,14 +137,12 @@ public class AMQ1925Test extends TestCase implements ExceptionListener {
// The runnable is likely to interrupt during the session#commit, since
// this takes the longest
final Object starter = new Object();
final CountDownLatch starter = new CountDownLatch(1);
final AtomicBoolean restarted = new AtomicBoolean();
new Thread(new Runnable() {
public void run() {
try {
synchronized (starter) {
starter.wait();
}
starter.await();
// Simulate broker failure & restart
bs.stop();
@ -167,9 +159,6 @@ public class AMQ1925Test extends TestCase implements ExceptionListener {
}
}).start();
synchronized (starter) {
starter.notifyAll();
}
Collection<Integer> results = new ArrayList<Integer>(MESSAGE_COUNT);
for (int i = 0; i < MESSAGE_COUNT; i++) {
Message message1 = consumer1.receive(20);
@ -191,9 +180,7 @@ public class AMQ1925Test extends TestCase implements ExceptionListener {
assertFalse("Timing problem, restarted too soon", restarted
.get());
if (i == 10) {
synchronized (starter) {
starter.notifyAll();
}
starter.countDown();
}
if (i > MESSAGE_COUNT - 50) {
assertTrue("Timing problem, restarted too late", restarted

6
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/udp/UdpTestSupport.java
View File

@ -46,7 +46,7 @@ public abstract class UdpTestSupport extends TestCase implements TransportListen
protected Transport producer;
protected Transport consumer;
protected Object lock = new Object();
protected final Object lock = new Object();
protected Command receivedCommand;
protected TransportServer server;
protected boolean large;
@ -251,10 +251,10 @@ public abstract class UdpTestSupport extends TestCase implements TransportListen
Command answer = null;
synchronized (lock) {
answer = receivedCommand;
if (answer == null) {
while (answer == null) {
lock.wait(waitForCommandTimeout);
answer = receivedCommand;
}
answer = receivedCommand;
}
assertNotNull("Should have received a Command by now!", answer);

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/usecases/BacklogNetworkCrossTalkTest.java
View File

@ -69,7 +69,7 @@ public class BacklogNetworkCrossTalkTest extends JmsMultipleBrokersTestSupport {
MessageConsumer clientB = createConsumer("B", destA);
final long maxWait = 5 * 60 * 1000l;
final long maxWait = 5 * 60 * 1000L;
MessageIdList listA = getConsumerMessages("A", clientA);
listA.setMaximumDuration(maxWait);
listA.waitForMessagesToArrive(numMessages);

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/usecases/BrokerQueueNetworkWithDisconnectTest.java
View File

@ -88,7 +88,7 @@ public class BrokerQueueNetworkWithDisconnectTest extends JmsMultipleBrokersTest
@SuppressWarnings("unchecked")
public void testNoStuckConnectionsWithTransportDisconnect() throws Exception {
inactiveDuration=60000l;
inactiveDuration=60000L;
useDuplexNetworkBridge = true;
bridgeBrokers(SPOKE, HUB);

16
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/usecases/ConcurrentProducerDurableConsumerTest.java
View File

@ -107,7 +107,7 @@ public class ConcurrentProducerDurableConsumerTest extends TestSupport {
// periodically start a durable sub that has a backlog
final int consumersToActivate = 5;
final Object addConsumerSignal = new Object();
final CountDownLatch addConsumerSignal = new CountDownLatch(1);
Executors.newCachedThreadPool(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
@ -120,9 +120,7 @@ public class ConcurrentProducerDurableConsumerTest extends TestSupport {
MessageConsumer consumer = null;
for (int i = 0; i < consumersToActivate; i++) {
LOG.info("Waiting for add signal from producer...");
synchronized (addConsumerSignal) {
addConsumerSignal.wait(30 * 60 * 1000);
}
addConsumerSignal.await(30, TimeUnit.MINUTES);
TimedMessageListener listener = new TimedMessageListener();
consumer = createDurableSubscriber(factory.createConnection(), destination, "consumer" + (i + 1));
LOG.info("Created consumer " + consumer);
@ -254,7 +252,7 @@ public class ConcurrentProducerDurableConsumerTest extends TestSupport {
final int numIterations,
Session session,
MessageProducer producer,
Object addConsumerSignal) throws Exception {
CountDownLatch addConsumerSignal) throws Exception {
long start;
long count = 0;
double batchMax = 0, max = 0, sum = 0;
@ -269,10 +267,8 @@ public class ConcurrentProducerDurableConsumerTest extends TestSupport {
max = Math.max(max, (System.currentTimeMillis() - singleSendstart));
if (++count % 500 == 0) {
if (addConsumerSignal != null) {
synchronized (addConsumerSignal) {
addConsumerSignal.notifyAll();
LOG.info("Signalled add consumer");
}
addConsumerSignal.countDown();
LOG.info("Signalled add consumer");
}
};
if (count % 5000 == 0) {
@ -405,7 +401,7 @@ public class ConcurrentProducerDurableConsumerTest extends TestSupport {
final int batchSize = 1000;
CountDownLatch firstReceiptLatch = new CountDownLatch(1);
long mark = System.currentTimeMillis();
long firstReceipt = 0l;
long firstReceipt = 0L;
long receiptAccumulator = 0;
long batchReceiptAccumulator = 0;
long maxReceiptTime = 0;

16
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/usecases/ConcurrentProducerQueueConsumerTest.java
View File

@ -95,7 +95,7 @@ public class ConcurrentProducerQueueConsumerTest extends TestSupport
// periodically start a queue consumer
final int consumersToActivate = 5;
final Object addConsumerSignal = new Object();
final CountDownLatch addConsumerSignal = new CountDownLatch(1);
Executors.newCachedThreadPool(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
@ -108,9 +108,7 @@ public class ConcurrentProducerQueueConsumerTest extends TestSupport
MessageConsumer consumer = null;
for (int i = 0; i < consumersToActivate; i++) {
LOG.info("Waiting for add signal from producer...");
synchronized (addConsumerSignal) {
addConsumerSignal.wait(30 * 60 * 1000);
}
addConsumerSignal.await(30, TimeUnit.MINUTES);
TimedMessageListener listener = new TimedMessageListener();
consumer = createConsumer(factory.createConnection(), destination);
LOG.info("Created consumer " + consumer);
@ -241,7 +239,7 @@ public class ConcurrentProducerQueueConsumerTest extends TestSupport
final int numIterations,
Session session,
MessageProducer producer,
Object addConsumerSignal) throws Exception {
CountDownLatch addConsumerSignal) throws Exception {
long start;
long count = 0;
double batchMax = 0, max = 0, sum = 0;
@ -257,10 +255,8 @@ public class ConcurrentProducerQueueConsumerTest extends TestSupport
max = Math.max(max, (System.currentTimeMillis() - singleSendstart));
if (++count % 500 == 0) {
if (addConsumerSignal != null) {
synchronized (addConsumerSignal) {
addConsumerSignal.notifyAll();
LOG.info("Signalled add consumer");
}
addConsumerSignal.countDown();
LOG.info("Signalled add consumer");
}
}
;
@ -362,7 +358,7 @@ public class ConcurrentProducerQueueConsumerTest extends TestSupport
final CountDownLatch firstReceiptLatch = new CountDownLatch(1);
long mark = System.currentTimeMillis();
long firstReceipt = 0l;
long firstReceipt = 0L;
long receiptAccumulator = 0;
long batchReceiptAccumulator = 0;
long maxReceiptTime = 0;

3
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubscriptionOffline4Test.java
View File

@ -30,6 +30,7 @@ import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
@ -80,7 +81,7 @@ public class DurableSubscriptionOffline4Test extends DurableSubscriptionOfflineT
MessageProducer producer = session.createProducer(null);
final int toSend = 500;
final String payload = new byte[40*1024].toString();
final String payload = Arrays.toString(new byte[40 * 1024]);
int sent = 0;
for (int i = sent; i < toSend; i++) {
Message message = session.createTextMessage(payload);

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/usecases/JdbcDurableSubDupTest.java
View File

@ -239,7 +239,7 @@ public class JdbcDurableSubDupTest {
ActiveMQConnectionFactory factory;
MessageProducer messageProducer;
long timeToLive = 0l;
long timeToLive = 0L;
TextMessage message = null;

4
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/usecases/MemoryLimitTest.java
View File

@ -168,7 +168,7 @@ public class MemoryLimitTest extends TestSupport {
final ProducerThread producer = new ProducerThread(sess, sess.createQueue("STORE.1")) {
@Override
protected Message createMessage(int i) throws Exception {
return sess.createTextMessage(payload + "::" + i);
return sess.createTextMessage(Arrays.toString(payload) + "::" + i);
}
};
producer.setMessageCount(1000);
@ -176,7 +176,7 @@ public class MemoryLimitTest extends TestSupport {
final ProducerThread producer2 = new ProducerThread(sess, sess.createQueue("STORE.2")) {
@Override
protected Message createMessage(int i) throws Exception {
return sess.createTextMessage(payload + "::" + i);
return sess.createTextMessage(Arrays.toString(payload) + "::" + i);
}
};
producer2.setMessageCount(1000);

4
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/usecases/MessageGroupReconnectDistributionTest.java
View File

@ -122,8 +122,8 @@ public class MessageGroupReconnectDistributionTest {
final ArrayList<AtomicLong> batchCounters = new ArrayList<AtomicLong>(numConsumers);
for (int i = 0; i < numConsumers; i++) {
consumedCounters.add(new AtomicLong(0l));
batchCounters.add(new AtomicLong(0l));
consumedCounters.add(new AtomicLong(0L));
batchCounters.add(new AtomicLong(0L));
final int id = i;
executorService.submit(new Runnable() {

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/usecases/MultiBrokersMultiClientsTest.java
View File

@ -44,7 +44,7 @@ public class MultiBrokersMultiClientsTest extends JmsMultipleBrokersTestSupport
private static final Logger LOG = LoggerFactory.getLogger(MultiBrokersMultiClientsTest.class);
protected Map<String, MessageConsumer> consumerMap;
Map<Thread, Throwable> unhandeledExceptions = new HashMap<Thread, Throwable>();
final Map<Thread, Throwable> unhandeledExceptions = new HashMap<Thread, Throwable>();
public void testTopicAllConnected() throws Exception {
bridgeAllBrokers();

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/usecases/NoDuplicateOnTopicNetworkTest.java
View File

@ -265,7 +265,7 @@ public class NoDuplicateOnTopicNetworkTest extends CombinationTestSupport {
private MessageConsumer consumer;
private final String durableID = "DURABLE_ID";
private List<String> receivedStrings = Collections.synchronizedList(new ArrayList<String>());
private final List<String> receivedStrings = Collections.synchronizedList(new ArrayList<String>());
private int numMessages = 10;
private CountDownLatch recievedLatch = new CountDownLatch(numMessages);

8
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/usecases/ReliableReconnectTest.java
View File

@ -47,8 +47,8 @@ public class ReliableReconnectTest extends org.apache.activemq.TestSupport {
protected int deliveryMode = DeliveryMode.PERSISTENT;
protected String consumerClientId;
protected Destination destination;
protected AtomicBoolean closeBroker = new AtomicBoolean(false);
protected AtomicInteger messagesReceived = new AtomicInteger(0);
protected final AtomicBoolean closeBroker = new AtomicBoolean(false);
protected final AtomicInteger messagesReceived = new AtomicInteger(0);
protected BrokerService broker;
protected int firstBatch = MESSAGE_COUNT / 10;
private IdGenerator idGen = new IdGenerator();
@ -159,7 +159,7 @@ public class ReliableReconnectTest extends org.apache.activemq.TestSupport {
connection.close();
spawnConsumer();
synchronized (closeBroker) {
if (!closeBroker.get()) {
while (!closeBroker.get()) {
closeBroker.wait();
}
}
@ -168,7 +168,7 @@ public class ReliableReconnectTest extends org.apache.activemq.TestSupport {
startBroker(false);
// System.err.println("Started Broker again");
synchronized (messagesReceived) {
if (messagesReceived.get() < MESSAGE_COUNT) {
while (messagesReceived.get() < MESSAGE_COUNT) {
messagesReceived.wait(60000);
}
}

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/usecases/ThreeBrokerTempDestDemandSubscriptionCleanupTest.java
View File

@ -122,7 +122,7 @@ public class ThreeBrokerTempDestDemandSubscriptionCleanupTest extends JmsMultipl
threadService.submit(tester);
threadService.shutdown();
assertTrue("executor done on time", threadService.awaitTermination(30l, TimeUnit.SECONDS));
assertTrue("executor done on time", threadService.awaitTermination(30L, TimeUnit.SECONDS));
// for the real test... we should not have any subscriptions left on broker C for the temp dests
BrokerItem brokerC = brokers.get(BROKER_C);

2
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/usecases/VerifyNetworkConsumersDisconnectTest.java
View File

@ -51,7 +51,7 @@ public class VerifyNetworkConsumersDisconnectTest extends JmsMultipleBrokersTest
public static final int TIMEOUT = 30000;
protected Map<String, MessageConsumer> consumerMap;
Map<Thread, Throwable> unhandledExceptions = new HashMap<Thread, Throwable>();
final Map<Thread, Throwable> unhandledExceptions = new HashMap<Thread, Throwable>();
private void assertNoUnhandledExceptions() {
for( Entry<Thread, Throwable> e: unhandledExceptions.entrySet()) {

32
tests/activemq5-unit-tests/src/test/java/org/apache/activemq/util/MessageIdList.java
View File

@ -140,20 +140,28 @@ public class MessageIdList extends Assert implements MessageListener {
long start = System.currentTimeMillis();
for (int i = 0; i < messageCount; i++) {
try {
if (hasReceivedMessages(messageCount)) {
break;
}
long duration = System.currentTimeMillis() - start;
if (duration >= maximumDuration) {
break;
}
synchronized (semaphore) {
synchronized (semaphore)
{
for (int i = 0; i < messageCount; i++)
{
try
{
if (hasReceivedMessages(messageCount))
{
break;
}
long duration = System.currentTimeMillis() - start;
if (duration >= maximumDuration)
{
break;
}
semaphore.wait(maximumDuration - duration);
}
} catch (InterruptedException e) {
LOG.info("Caught: " + e);
catch (InterruptedException e)
{
LOG.info("Caught: " + e);
}
}
}
long end = System.currentTimeMillis() - start;