mirror of https://github.com/apache/activemq.git
fix test cases after changes in https://issues.apache.org/jira/browse/AMQ-4237 broker the tests queue MBean lookup
git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1427878 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
64f3c55d80
commit
19bf943dd7
|
@ -55,102 +55,102 @@ import org.slf4j.LoggerFactory;
|
|||
|
||||
/**
|
||||
* Modified CursorSupport Unit test to reproduce the negative queue issue.
|
||||
*
|
||||
*
|
||||
* Keys to reproducing:
|
||||
* 1) Consecutive queues with listener on first sending to second queue
|
||||
* 2) Push each queue to the memory limit
|
||||
* This seems to help reproduce the issue more consistently, but
|
||||
* we have seen times in our production environment where the
|
||||
* negative queue can occur without. Our memory limits are
|
||||
* very high in production and it still happens in varying
|
||||
* very high in production and it still happens in varying
|
||||
* frequency.
|
||||
* 3) Prefetch
|
||||
* Lowering the prefetch down to 10 and below seems to help
|
||||
* reduce occurrences.
|
||||
* Lowering the prefetch down to 10 and below seems to help
|
||||
* reduce occurrences.
|
||||
* 4) # of consumers per queue
|
||||
* The issue occurs less with fewer consumers
|
||||
*
|
||||
*
|
||||
* Things that do not affect reproduction:
|
||||
* 1) Spring - we use spring in our production applications, but this test case works
|
||||
* with or without it.
|
||||
* 2) transacted
|
||||
*
|
||||
*
|
||||
*/
|
||||
public class NegativeQueueTest extends AutoFailTestSupport {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(NegativeQueueTest.class);
|
||||
|
||||
|
||||
public static SimpleDateFormat formatter = new SimpleDateFormat("yyyyMMdd,hh:mm:ss:SSS");
|
||||
|
||||
|
||||
private static final String QUEUE_1_NAME = "conn.test.queue.1";
|
||||
private static final String QUEUE_2_NAME = "conn.test.queue.2";
|
||||
|
||||
|
||||
private static final long QUEUE_MEMORY_LIMIT = 2097152;
|
||||
private static final long MEMORY_USAGE = 400000000;
|
||||
private static final long TEMP_USAGE = 200000000;
|
||||
private static final long STORE_USAGE = 1000000000;
|
||||
private static final int MESSAGE_COUNT = 1100;
|
||||
|
||||
|
||||
protected static final boolean TRANSACTED = true;
|
||||
protected static final boolean DEBUG = true;
|
||||
protected static int NUM_CONSUMERS = 20;
|
||||
protected static int PREFETCH_SIZE = 1000;
|
||||
|
||||
protected static int NUM_CONSUMERS = 20;
|
||||
protected static int PREFETCH_SIZE = 1000;
|
||||
|
||||
protected BrokerService broker;
|
||||
protected String bindAddress = "tcp://localhost:0";
|
||||
|
||||
|
||||
public void testWithDefaultPrefetch() throws Exception{
|
||||
PREFETCH_SIZE = 1000;
|
||||
NUM_CONSUMERS = 20;
|
||||
blastAndConsume();
|
||||
}
|
||||
|
||||
|
||||
public void x_testWithDefaultPrefetchFiveConsumers() throws Exception{
|
||||
PREFETCH_SIZE = 1000;
|
||||
NUM_CONSUMERS = 5;
|
||||
blastAndConsume();
|
||||
}
|
||||
|
||||
|
||||
public void x_testWithDefaultPrefetchTwoConsumers() throws Exception{
|
||||
PREFETCH_SIZE = 1000;
|
||||
NUM_CONSUMERS = 2;
|
||||
blastAndConsume();
|
||||
}
|
||||
|
||||
|
||||
public void testWithDefaultPrefetchOneConsumer() throws Exception{
|
||||
PREFETCH_SIZE = 1000;
|
||||
NUM_CONSUMERS = 1;
|
||||
blastAndConsume();
|
||||
}
|
||||
|
||||
|
||||
public void testWithMediumPrefetch() throws Exception{
|
||||
PREFETCH_SIZE = 50;
|
||||
NUM_CONSUMERS = 20;
|
||||
blastAndConsume();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public void x_testWithSmallPrefetch() throws Exception{
|
||||
PREFETCH_SIZE = 10;
|
||||
NUM_CONSUMERS = 20;
|
||||
blastAndConsume();
|
||||
}
|
||||
|
||||
|
||||
public void testWithNoPrefetch() throws Exception{
|
||||
PREFETCH_SIZE = 1;
|
||||
NUM_CONSUMERS = 20;
|
||||
blastAndConsume();
|
||||
}
|
||||
|
||||
|
||||
public void blastAndConsume() throws Exception {
|
||||
LOG.info(getName());
|
||||
ConnectionFactory factory = createConnectionFactory();
|
||||
|
||||
|
||||
//get proxy queues for statistics lookups
|
||||
Connection proxyConnection = factory.createConnection();
|
||||
proxyConnection.start();
|
||||
Session proxySession = proxyConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
|
||||
final QueueViewMBean proxyQueue1 = getProxyToQueueViewMBean((Queue)proxySession.createQueue(QUEUE_1_NAME));
|
||||
final QueueViewMBean proxyQueue2 = getProxyToQueueViewMBean((Queue)proxySession.createQueue(QUEUE_2_NAME));
|
||||
|
||||
final QueueViewMBean proxyQueue1 = getProxyToQueueViewMBean(proxySession.createQueue(QUEUE_1_NAME));
|
||||
final QueueViewMBean proxyQueue2 = getProxyToQueueViewMBean(proxySession.createQueue(QUEUE_2_NAME));
|
||||
|
||||
// LOAD THE QUEUE
|
||||
Connection producerConnection = factory.createConnection();
|
||||
producerConnection.start();
|
||||
|
@ -168,7 +168,7 @@ public class NegativeQueueTest extends AutoFailTestSupport {
|
|||
System.out.print(index-((index/10)*10));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
//get access to the Queue info
|
||||
if(DEBUG){
|
||||
System.out.println("");
|
||||
|
@ -176,16 +176,16 @@ public class NegativeQueueTest extends AutoFailTestSupport {
|
|||
System.out.println("Queue1 Memory % Used = "+proxyQueue1.getMemoryPercentUsage());
|
||||
System.out.println("Queue1 Memory Available = "+proxyQueue1.getMemoryLimit());
|
||||
}
|
||||
|
||||
|
||||
// FLUSH THE QUEUE
|
||||
final CountDownLatch latch1 = new CountDownLatch(1);
|
||||
final CountDownLatch latch2 = new CountDownLatch(1);
|
||||
final CountDownLatch latch2 = new CountDownLatch(1);
|
||||
Connection[] consumerConnections1 = new Connection[NUM_CONSUMERS];
|
||||
List<Message> consumerList1 = new ArrayList<Message>();
|
||||
Connection[] consumerConnections2 = new Connection[NUM_CONSUMERS];
|
||||
Connection[] producerConnections2 = new Connection[NUM_CONSUMERS];
|
||||
List<Message> consumerList2 = new ArrayList<Message>();
|
||||
|
||||
|
||||
for(int ix=0; ix<NUM_CONSUMERS; ix++){
|
||||
producerConnections2[ix] = factory.createConnection();
|
||||
producerConnections2[ix].start();
|
||||
|
@ -194,7 +194,7 @@ public class NegativeQueueTest extends AutoFailTestSupport {
|
|||
MessageConsumer consumer = consumerSession.createConsumer(session.createQueue(QUEUE_1_NAME));
|
||||
consumer.setMessageListener(new SessionAwareMessageListener(producerConnections2[ix], consumerSession, QUEUE_2_NAME, latch1, consumerList1));
|
||||
}
|
||||
|
||||
|
||||
latch1.await(200000, TimeUnit.MILLISECONDS);
|
||||
if(DEBUG){
|
||||
System.out.println("");
|
||||
|
@ -202,15 +202,16 @@ public class NegativeQueueTest extends AutoFailTestSupport {
|
|||
System.out.println("Queue2 Memory % Used = "+proxyQueue2.getMemoryPercentUsage());
|
||||
System.out.println("Queue2 Memory Available = "+proxyQueue2.getMemoryLimit());
|
||||
}
|
||||
|
||||
|
||||
for(int ix=0; ix<NUM_CONSUMERS; ix++){
|
||||
consumerConnections2[ix] = getConsumerConnection(factory);
|
||||
Session consumerSession = consumerConnections2[ix].createSession(TRANSACTED, Session.AUTO_ACKNOWLEDGE);
|
||||
MessageConsumer consumer = consumerSession.createConsumer(session.createQueue(QUEUE_2_NAME));
|
||||
consumer.setMessageListener(new SessionAwareMessageListener(consumerSession, latch2, consumerList2));
|
||||
}
|
||||
|
||||
|
||||
boolean success = Wait.waitFor(new Wait.Condition() {
|
||||
@Override
|
||||
public boolean isSatisified() throws Exception {
|
||||
boolean done = latch2.await(10, TimeUnit.SECONDS);
|
||||
if(DEBUG){
|
||||
|
@ -235,10 +236,10 @@ public class NegativeQueueTest extends AutoFailTestSupport {
|
|||
consumerConnections2[ix].close();
|
||||
producerConnections2[ix].close();
|
||||
}
|
||||
|
||||
|
||||
//let the consumer statistics on queue2 have time to update
|
||||
Thread.sleep(500);
|
||||
|
||||
|
||||
if(DEBUG){
|
||||
System.out.println("");
|
||||
System.out.println("Queue1 Size = "+proxyQueue1.getQueueSize());
|
||||
|
@ -248,37 +249,39 @@ public class NegativeQueueTest extends AutoFailTestSupport {
|
|||
}
|
||||
|
||||
Wait.waitFor(new Wait.Condition() {
|
||||
@Override
|
||||
public boolean isSatisified() throws Exception {
|
||||
return 0 == proxyQueue1.getQueueSize();
|
||||
}});
|
||||
assertEquals("Queue1 has gone negative,",0, proxyQueue1.getQueueSize());
|
||||
|
||||
|
||||
Wait.waitFor(new Wait.Condition() {
|
||||
@Override
|
||||
public boolean isSatisified() throws Exception {
|
||||
return 0 == proxyQueue2.getQueueSize();
|
||||
}});
|
||||
assertEquals("Queue2 has gone negative,",0, proxyQueue2.getQueueSize());
|
||||
proxyConnection.close();
|
||||
|
||||
|
||||
}
|
||||
|
||||
private QueueViewMBean getProxyToQueueViewMBean(Queue queue)
|
||||
throws MalformedObjectNameException, JMSException {
|
||||
|
||||
ObjectName queueViewMBeanName = new ObjectName("org.apache.activemq" + ":Type=Queue,Destination=" +
|
||||
queue.getQueueName() + ",BrokerName=localhost");
|
||||
QueueViewMBean proxy = (QueueViewMBean) broker.getManagementContext().newProxyInstance(queueViewMBeanName,
|
||||
QueueViewMBean.class, true);
|
||||
|
||||
private QueueViewMBean getProxyToQueueViewMBean(Queue queue) throws MalformedObjectNameException, JMSException {
|
||||
final String prefix = "org.apache.activemq:type=Broker,brokerName=localhost,destinationType=Queue,destinationName=";
|
||||
|
||||
ObjectName queueViewMBeanName = new ObjectName(prefix + queue.getQueueName());
|
||||
QueueViewMBean proxy = (QueueViewMBean)
|
||||
broker.getManagementContext().newProxyInstance(queueViewMBeanName, QueueViewMBean.class, true);
|
||||
|
||||
return proxy;
|
||||
}
|
||||
|
||||
|
||||
protected Connection getConsumerConnection(ConnectionFactory fac) throws JMSException {
|
||||
Connection connection = fac.createConnection();
|
||||
connection.start();
|
||||
return connection;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void setUp() throws Exception {
|
||||
if (broker == null) {
|
||||
broker = createBroker();
|
||||
|
@ -286,6 +289,7 @@ public class NegativeQueueTest extends AutoFailTestSupport {
|
|||
super.setUp();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void tearDown() throws Exception {
|
||||
super.tearDown();
|
||||
if (broker != null) {
|
||||
|
@ -312,18 +316,18 @@ public class NegativeQueueTest extends AutoFailTestSupport {
|
|||
bindAddress = answer.getTransportConnectors().get(0).getConnectUri().toString();
|
||||
return answer;
|
||||
}
|
||||
|
||||
|
||||
protected void configureBroker(BrokerService answer) throws Exception {
|
||||
PolicyEntry policy = new PolicyEntry();
|
||||
policy.setMemoryLimit(QUEUE_MEMORY_LIMIT);
|
||||
policy.setMemoryLimit(QUEUE_MEMORY_LIMIT);
|
||||
policy.setPendingQueuePolicy(new StorePendingQueueMessageStoragePolicy());
|
||||
|
||||
|
||||
// disable the cache to be sure setBatch is the problem
|
||||
// will get lots of duplicates
|
||||
// real problem is sync between cursor and store add - leads to out or order messages
|
||||
// in the cursor so setBatch can break.
|
||||
// policy.setUseCache(false);
|
||||
|
||||
|
||||
PolicyMap pMap = new PolicyMap();
|
||||
pMap.setDefaultEntry(policy);
|
||||
answer.setDestinationPolicy(pMap);
|
||||
|
@ -331,14 +335,14 @@ public class NegativeQueueTest extends AutoFailTestSupport {
|
|||
answer.addConnector("tcp://localhost:0");
|
||||
|
||||
MemoryUsage memoryUsage = new MemoryUsage();
|
||||
memoryUsage.setLimit(MEMORY_USAGE);
|
||||
memoryUsage.setLimit(MEMORY_USAGE);
|
||||
memoryUsage.setPercentUsageMinDelta(20);
|
||||
|
||||
TempUsage tempUsage = new TempUsage();
|
||||
tempUsage.setLimit(TEMP_USAGE);
|
||||
tempUsage.setLimit(TEMP_USAGE);
|
||||
|
||||
StoreUsage storeUsage = new StoreUsage();
|
||||
storeUsage.setLimit(STORE_USAGE);
|
||||
storeUsage.setLimit(STORE_USAGE);
|
||||
|
||||
SystemUsage systemUsage = new SystemUsage();
|
||||
systemUsage.setMemoryUsage(memoryUsage);
|
||||
|
@ -346,27 +350,27 @@ public class NegativeQueueTest extends AutoFailTestSupport {
|
|||
systemUsage.setStoreUsage(storeUsage);
|
||||
answer.setSystemUsage(systemUsage);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Message listener that is given the Session for transacted consumers
|
||||
*/
|
||||
class SessionAwareMessageListener implements MessageListener{
|
||||
private List<Message> consumerList;
|
||||
private CountDownLatch latch;
|
||||
private Session consumerSession;
|
||||
private final List<Message> consumerList;
|
||||
private final CountDownLatch latch;
|
||||
private final Session consumerSession;
|
||||
private Session producerSession;
|
||||
private MessageProducer producer;
|
||||
|
||||
|
||||
public SessionAwareMessageListener(Session consumerSession, CountDownLatch latch, List<Message> consumerList){
|
||||
this(null, consumerSession, null, latch, consumerList);
|
||||
}
|
||||
|
||||
public SessionAwareMessageListener(Connection producerConnection, Session consumerSession, String outQueueName,
|
||||
|
||||
public SessionAwareMessageListener(Connection producerConnection, Session consumerSession, String outQueueName,
|
||||
CountDownLatch latch, List<Message> consumerList){
|
||||
this.consumerList = consumerList;
|
||||
this.latch = latch;
|
||||
this.consumerSession = consumerSession;
|
||||
|
||||
|
||||
if(producerConnection != null){
|
||||
try {
|
||||
producerSession = producerConnection.createSession(TRANSACTED, Session.AUTO_ACKNOWLEDGE);
|
||||
|
@ -377,7 +381,8 @@ public class NegativeQueueTest extends AutoFailTestSupport {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public void onMessage(Message msg) {
|
||||
try {
|
||||
if(producer == null){
|
||||
|
@ -392,7 +397,7 @@ public class NegativeQueueTest extends AutoFailTestSupport {
|
|||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
|
||||
|
||||
synchronized(consumerList){
|
||||
consumerList.add(msg);
|
||||
if(DEBUG && consumerList.size()%100 == 0) {
|
||||
|
@ -411,5 +416,5 @@ public class NegativeQueueTest extends AutoFailTestSupport {
|
|||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
@ -98,7 +98,8 @@ public class AMQ2513Test extends TestCase {
|
|||
|
||||
DestinationViewMBean createView() throws Exception {
|
||||
String domain = "org.apache.activemq";
|
||||
ObjectName name = new ObjectName(domain + ":BrokerName=localhost,Type=Queue,Destination=test");
|
||||
ObjectName name = new ObjectName(domain + ":type=Broker,brokerName=localhost," +
|
||||
"destinationType=Queue,destinationName=test");
|
||||
return (DestinationViewMBean) broker.getManagementContext().newProxyInstance(name, DestinationViewMBean.class,
|
||||
true);
|
||||
}
|
||||
|
|
|
@ -163,10 +163,6 @@ public class ExpiredMessagesWithNoConsumerTest extends CombinationTestSupport {
|
|||
|
||||
TimeUnit.SECONDS.sleep(5);
|
||||
|
||||
for (ObjectName name : broker.getAdminView().getQueues()) {
|
||||
LOG.info("Broker Queue: {}", name);
|
||||
}
|
||||
|
||||
final DestinationViewMBean view = createView(destination);
|
||||
Wait.waitFor(new Wait.Condition() {
|
||||
@Override
|
||||
|
@ -581,8 +577,6 @@ public class ExpiredMessagesWithNoConsumerTest extends CombinationTestSupport {
|
|||
name = new ObjectName(domain + ":type=Broker,brokerName=localhost,destinationType=Topic,destinationName=test");
|
||||
}
|
||||
|
||||
LOG.info("Attempting to find Queue named: {}", name);
|
||||
|
||||
return (DestinationViewMBean) broker.getManagementContext().newProxyInstance(name, DestinationViewMBean.class, true);
|
||||
}
|
||||
|
||||
|
|
Loading…
Reference in New Issue