Clean up the test, use timeouts and add a few more iterations to try and
reproduce recent CI failures from this test.
This commit is contained in:
Timothy Bish 2016-04-13 15:17:14 -04:00
parent 28e4090d95
commit 82330793e9
1 changed files with 52 additions and 86 deletions

View File

@ -1,4 +1,4 @@
/** /*
* Licensed to the Apache Software Foundation (ASF) under one or more * Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with * contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership. * this work for additional information regarding copyright ownership.
@ -16,9 +16,14 @@
*/ */
package org.apache.activemq.usecases; package org.apache.activemq.usecases;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Enumeration; import java.util.Enumeration;
import java.util.List; import java.util.List;
import javax.jms.BytesMessage; import javax.jms.BytesMessage;
import javax.jms.Connection; import javax.jms.Connection;
import javax.jms.ConnectionFactory; import javax.jms.ConnectionFactory;
@ -27,62 +32,60 @@ import javax.jms.MessageConsumer;
import javax.jms.MessageProducer; import javax.jms.MessageProducer;
import javax.jms.QueueBrowser; import javax.jms.QueueBrowser;
import javax.jms.Session; import javax.jms.Session;
import junit.framework.TestCase;
import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.region.policy.PolicyEntry; import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap; import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.command.ActiveMQQueue; import org.apache.activemq.command.ActiveMQQueue;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
public class QueueZeroPrefetchLazyDispatchPriorityTest {
public class QueueZeroPrefetchLazyDispatchPriorityTest extends TestCase { private static final Logger LOG = LoggerFactory.getLogger(QueueZeroPrefetchLazyDispatchPriorityTest.class);
private final byte[] PAYLOAD = new byte[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 };
private final int ITERATIONS = 10;
static final Logger LOG = LoggerFactory.getLogger(QueueZeroPrefetchLazyDispatchPriorityTest.class);
private BrokerService broker; private BrokerService broker;
public static final byte[] PAYLOAD = new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9};
@Before
protected void setUp() throws Exception { public void setUp() throws Exception {
broker = createBroker(); broker = createBroker();
broker.start(); broker.start();
broker.waitUntilStarted(); broker.waitUntilStarted();
} }
protected void tearDown() throws Exception { @After
public void tearDown() throws Exception {
if (broker != null) { if (broker != null) {
broker.stop(); broker.stop();
} }
} }
@Test(timeout=90000)
public void testPriorityMessages() throws Exception { public void testPriorityMessages() throws Exception {
for (int i = 0; i < ITERATIONS; i++) {
for (int i = 0; i < 5; i++) { // send 4 message priority MEDIUM
//send 4 message priority MEDIUM
produceMessages(4, 4, "TestQ"); produceMessages(4, 4, "TestQ");
// send 1 message priority HIGH
//send 1 message priority HIGH
produceMessages(1, 5, "TestQ"); produceMessages(1, 5, "TestQ");
LOG.info("On iteration {}", i);
LOG.info("On iteration " + i);
Thread.sleep(500); Thread.sleep(500);
// consume messages // consume messages
ArrayList<Message> consumeList = consumeMessages("TestQ"); ArrayList<Message> consumeList = consumeMessages("TestQ");
LOG.info("Consumed list " + consumeList.size()); LOG.info("Consumed list " + consumeList.size());
// compare lists // compare lists
assertEquals("message 1 should be priority high", 5, consumeList.get(0).getJMSPriority()); assertEquals("message 1 should be priority high", 5, consumeList.get(0).getJMSPriority());
assertEquals("message 2 should be priority medium", 4, consumeList.get(1).getJMSPriority()); assertEquals("message 2 should be priority medium", 4, consumeList.get(1).getJMSPriority());
@ -90,99 +93,91 @@ public class QueueZeroPrefetchLazyDispatchPriorityTest extends TestCase {
assertEquals("message 4 should be priority medium", 4, consumeList.get(3).getJMSPriority()); assertEquals("message 4 should be priority medium", 4, consumeList.get(3).getJMSPriority());
assertEquals("message 5 should be priority medium", 4, consumeList.get(4).getJMSPriority()); assertEquals("message 5 should be priority medium", 4, consumeList.get(4).getJMSPriority());
} }
} }
@Test(timeout=120000)
public void testPriorityMessagesMoreThanPageSize() throws Exception { public void testPriorityMessagesMoreThanPageSize() throws Exception {
final int numToSend = 450; final int numToSend = 450;
for (int i = 0; i < 5; i++) { for (int i = 0; i < ITERATIONS; i++) {
produceMessages(numToSend - 1, 4, "TestQ"); produceMessages(numToSend - 1, 4, "TestQ");
// ensure we get expiry processing // ensure we get expiry processing
Thread.sleep(700); Thread.sleep(700);
// send 1 message priority HIGH
//send 1 message priority HIGH
produceMessages(1, 5, "TestQ"); produceMessages(1, 5, "TestQ");
Thread.sleep(500); Thread.sleep(500);
LOG.info("On iteration " + i); LOG.info("On iteration {}", i);
// consume messages // consume messages
ArrayList<Message> consumeList = consumeMessages("TestQ"); ArrayList<Message> consumeList = consumeMessages("TestQ");
LOG.info("Consumed list " + consumeList.size()); LOG.info("Consumed list {}", consumeList.size());
// compare lists // compare lists
assertFalse("Consumed list should not be empty", consumeList.isEmpty());
assertEquals("message 1 should be priority high", 5, consumeList.get(0).getJMSPriority()); assertEquals("message 1 should be priority high", 5, consumeList.get(0).getJMSPriority());
for (int j = 1; j < (numToSend - 1); j++) { for (int j = 1; j < (numToSend - 1); j++) {
assertEquals("message " + j + " should be priority medium", 4, consumeList.get(j).getJMSPriority()); assertEquals("message " + j + " should be priority medium", 4, consumeList.get(j).getJMSPriority());
} }
} }
} }
@Test(timeout=90000)
public void testLongLivedPriorityConsumer() throws Exception { public void testLongLivedPriorityConsumer() throws Exception {
final int numToSend = 150; final int numToSend = 150;
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(broker.getTransportConnectorByScheme("tcp").getPublishableConnectString()); ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(broker.getTransportConnectorByScheme("tcp").getPublishableConnectString());
Connection connection = connectionFactory.createConnection(); Connection connection = connectionFactory.createConnection();
try { try {
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
MessageConsumer consumer = session.createConsumer(new ActiveMQQueue("TestQ")); MessageConsumer consumer = session.createConsumer(new ActiveMQQueue("TestQ"));
connection.start(); connection.start();
for (int i = 0; i < 5; i++) { for (int i = 0; i < ITERATIONS; i++) {
produceMessages(numToSend - 1, 4, "TestQ"); produceMessages(numToSend - 1, 4, "TestQ");
//send 1 message priority HIGH // send 1 message priority HIGH
produceMessages(1, 5, "TestQ"); produceMessages(1, 5, "TestQ");
Message message = consumer.receive(4000); Message message = consumer.receive(4000);
assertEquals("message should be priority high", 5, message.getJMSPriority()); assertEquals("message should be priority high", 5, message.getJMSPriority());
} }
} finally { } finally {
connection.close(); connection.close();
} }
ArrayList<Message> consumeList = consumeMessages("TestQ"); ArrayList<Message> consumeList = consumeMessages("TestQ");
LOG.info("Consumed list " + consumeList.size()); LOG.info("Consumed list {}", consumeList.size());
for (Message message : consumeList) { for (Message message : consumeList) {
assertEquals("should be priority medium", 4, message.getJMSPriority()); assertEquals("should be priority medium", 4, message.getJMSPriority());
} }
} }
@Test(timeout=90000)
public void testPriorityMessagesWithJmsBrowser() throws Exception { public void testPriorityMessagesWithJmsBrowser() throws Exception {
final int numToSend = 250; final int numToSend = 250;
for (int i = 0; i < 5; i++) {
for (int i = 0; i < ITERATIONS; i++) {
produceMessages(numToSend - 1, 4, "TestQ"); produceMessages(numToSend - 1, 4, "TestQ");
ArrayList<Message> browsed = browseMessages("TestQ"); ArrayList<Message> browsed = browseMessages("TestQ");
LOG.info("Browsed: " + browsed.size()); LOG.info("Browsed: {}", browsed.size());
//send 1 message priority HIGH // send 1 message priority HIGH
produceMessages(1, 5, "TestQ"); produceMessages(1, 5, "TestQ");
Thread.sleep(500); Thread.sleep(500);
LOG.info("On iteration " + i); LOG.info("On iteration {}", i);
Message message = consumeOneMessage("TestQ"); Message message = consumeOneMessage("TestQ");
assertNotNull(message); assertNotNull(message);
@ -190,29 +185,28 @@ public class QueueZeroPrefetchLazyDispatchPriorityTest extends TestCase {
// consume messages // consume messages
ArrayList<Message> consumeList = consumeMessages("TestQ"); ArrayList<Message> consumeList = consumeMessages("TestQ");
LOG.info("Consumed list " + consumeList.size()); LOG.info("Consumed list {}", consumeList.size());
// compare lists // compare lists
//assertEquals("Iteration: " + i +", message 1 should be priority high", 5, consumeList.get(0).getJMSPriority()); // assertEquals("Iteration: " + i
// +", message 1 should be priority high", 5,
// consumeList.get(0).getJMSPriority());
for (int j = 1; j < (numToSend - 1); j++) { for (int j = 1; j < (numToSend - 1); j++) {
assertEquals("Iteration: " + i + ", message " + j + " should be priority medium", 4, consumeList.get(j).getJMSPriority()); assertEquals("Iteration: " + i + ", message " + j + " should be priority medium", 4, consumeList.get(j).getJMSPriority());
} }
} }
} }
@Test(timeout=90000)
public void testJmsBrowserGetsPagedIn() throws Exception { public void testJmsBrowserGetsPagedIn() throws Exception {
final int numToSend = 10; final int numToSend = 10;
for (int i = 0; i < 10; i++) {
for (int i = 0; i < ITERATIONS; i++) {
produceMessages(numToSend, 4, "TestQ"); produceMessages(numToSend, 4, "TestQ");
ArrayList<Message> browsed = browseMessages("TestQ"); ArrayList<Message> browsed = browseMessages("TestQ");
LOG.info("Browsed: " + browsed.size()); LOG.info("Browsed: {}", browsed.size());
assertEquals(0, browsed.size()); assertEquals(0, browsed.size());
@ -221,7 +215,7 @@ public class QueueZeroPrefetchLazyDispatchPriorityTest extends TestCase {
browsed = browseMessages("TestQ"); browsed = browseMessages("TestQ");
LOG.info("Browsed: " + browsed.size()); LOG.info("Browsed: {}", browsed.size());
assertEquals("see only the paged in for pull", 1, browsed.size()); assertEquals("see only the paged in for pull", 1, browsed.size());
@ -229,41 +223,33 @@ public class QueueZeroPrefetchLazyDispatchPriorityTest extends TestCase {
ArrayList<Message> consumeList = consumeMessages("TestQ"); ArrayList<Message> consumeList = consumeMessages("TestQ");
LOG.info("Consumed list " + consumeList.size()); LOG.info("Consumed list " + consumeList.size());
assertEquals(numToSend, consumeList.size()); assertEquals(numToSend, consumeList.size());
} }
} }
private void produceMessages(int numberOfMessages, int priority, String queueName) throws Exception { private void produceMessages(int numberOfMessages, int priority, String queueName) throws Exception {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(broker.getTransportConnectorByScheme("tcp").getPublishableConnectString()); ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(broker.getTransportConnectorByScheme("tcp").getPublishableConnectString());
connectionFactory.setConnectionIDPrefix("pri-" + priority); connectionFactory.setConnectionIDPrefix("pri-" + priority);
Connection connection = connectionFactory.createConnection(); Connection connection = connectionFactory.createConnection();
try { try {
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
MessageProducer producer = session.createProducer(new ActiveMQQueue(queueName)); MessageProducer producer = session.createProducer(new ActiveMQQueue(queueName));
connection.start(); connection.start();
for (int i = 0; i < numberOfMessages; i++) { for (int i = 0; i < numberOfMessages; i++) {
BytesMessage m = session.createBytesMessage(); BytesMessage m = session.createBytesMessage();
m.writeBytes(PAYLOAD); m.writeBytes(PAYLOAD);
m.setJMSPriority(priority); m.setJMSPriority(priority);
producer.send(m, Message.DEFAULT_DELIVERY_MODE, m.getJMSPriority(), Message.DEFAULT_TIME_TO_LIVE); producer.send(m, Message.DEFAULT_DELIVERY_MODE, m.getJMSPriority(), Message.DEFAULT_TIME_TO_LIVE);
} }
} finally { } finally {
if (connection != null) { if (connection != null) {
connection.close(); connection.close();
} }
} }
} }
private ArrayList<Message> consumeMessages(String queueName) throws Exception { private ArrayList<Message> consumeMessages(String queueName) throws Exception {
ArrayList<Message> returnedMessages = new ArrayList<Message>(); ArrayList<Message> returnedMessages = new ArrayList<Message>();
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(broker.getTransportConnectorByScheme("tcp").getPublishableConnectString()); ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(broker.getTransportConnectorByScheme("tcp").getPublishableConnectString());
@ -275,7 +261,6 @@ public class QueueZeroPrefetchLazyDispatchPriorityTest extends TestCase {
boolean finished = false; boolean finished = false;
while (!finished) { while (!finished) {
Message message = consumer.receive(1000); Message message = consumer.receive(1000);
if (message == null) { if (message == null) {
finished = true; finished = true;
@ -284,20 +269,15 @@ public class QueueZeroPrefetchLazyDispatchPriorityTest extends TestCase {
if (message != null) { if (message != null) {
returnedMessages.add(message); returnedMessages.add(message);
} }
} }
consumer.close(); consumer.close();
return returnedMessages; return returnedMessages;
} finally { } finally {
if (connection != null) { if (connection != null) {
connection.close(); connection.close();
} }
} }
} }
private Message consumeOneMessage(String queueName) throws Exception { private Message consumeOneMessage(String queueName) throws Exception {
@ -305,7 +285,6 @@ public class QueueZeroPrefetchLazyDispatchPriorityTest extends TestCase {
} }
private Message consumeOneMessage(String queueName, int ackMode) throws Exception { private Message consumeOneMessage(String queueName, int ackMode) throws Exception {
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(broker.getTransportConnectorByScheme("tcp").getPublishableConnectString()); ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(broker.getTransportConnectorByScheme("tcp").getPublishableConnectString());
Connection connection = connectionFactory.createConnection(); Connection connection = connectionFactory.createConnection();
try { try {
@ -313,19 +292,15 @@ public class QueueZeroPrefetchLazyDispatchPriorityTest extends TestCase {
MessageConsumer consumer = session.createConsumer(new ActiveMQQueue(queueName)); MessageConsumer consumer = session.createConsumer(new ActiveMQQueue(queueName));
connection.start(); connection.start();
return consumer.receive(1000); return consumer.receive(2000);
} finally { } finally {
if (connection != null) { if (connection != null) {
connection.close(); connection.close();
} }
} }
} }
private ArrayList<Message> browseMessages(String queueName) throws Exception { private ArrayList<Message> browseMessages(String queueName) throws Exception {
ArrayList<Message> returnedMessages = new ArrayList<Message>(); ArrayList<Message> returnedMessages = new ArrayList<Message>();
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(broker.getTransportConnectorByScheme("tcp").getPublishableConnectString()); ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(broker.getTransportConnectorByScheme("tcp").getPublishableConnectString());
@ -335,31 +310,25 @@ public class QueueZeroPrefetchLazyDispatchPriorityTest extends TestCase {
QueueBrowser consumer = session.createBrowser(new ActiveMQQueue(queueName)); QueueBrowser consumer = session.createBrowser(new ActiveMQQueue(queueName));
connection.start(); connection.start();
Enumeration enumeration = consumer.getEnumeration(); Enumeration<?> enumeration = consumer.getEnumeration();
while (enumeration.hasMoreElements()) { while (enumeration.hasMoreElements()) {
Message message = (Message) enumeration.nextElement(); Message message = (Message) enumeration.nextElement();
returnedMessages.add(message); returnedMessages.add(message);
} }
return returnedMessages; return returnedMessages;
} finally { } finally {
if (connection != null) { if (connection != null) {
connection.close(); connection.close();
} }
} }
} }
private BrokerService createBroker() throws Exception { private BrokerService createBroker() throws Exception {
BrokerService broker = new BrokerService(); BrokerService broker = new BrokerService();
broker.setDeleteAllMessagesOnStartup(true); broker.setDeleteAllMessagesOnStartup(true);
//add the policy entries // add the policy entries
PolicyMap policyMap = new PolicyMap(); PolicyMap policyMap = new PolicyMap();
List<PolicyEntry> entries = new ArrayList<PolicyEntry>(); List<PolicyEntry> entries = new ArrayList<PolicyEntry>();
PolicyEntry pe = new PolicyEntry(); PolicyEntry pe = new PolicyEntry();
@ -384,10 +353,7 @@ public class QueueZeroPrefetchLazyDispatchPriorityTest extends TestCase {
policyMap.setPolicyEntries(entries); policyMap.setPolicyEntries(entries);
broker.setDestinationPolicy(policyMap); broker.setDestinationPolicy(policyMap);
broker.addConnector("tcp://0.0.0.0:0"); broker.addConnector("tcp://0.0.0.0:0");
return broker; return broker;
} }
} }