From 73b7ec00e546b27d2bed22d30ad48a8735cfe6f0 Mon Sep 17 00:00:00 2001 From: Alan Conway Date: Mon, 28 Mar 2011 23:18:16 +0000 Subject: Merge branch 'trunk' into qpid-2920 Conflicts: qpid/cpp/src/cluster.mk qpid/java/systests/src/main/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/qpid-2920@1086439 13f79535-47bb-0310-9956-ffa450edef68 --- .../apache/qpid/server/queue/SimpleAMQQueue.java | 91 ++++++-- .../qpid/server/transport/ServerConnection.java | 14 +- .../java/org/apache/qpid/client/AMQSession.java | 5 + .../qpid/client/BasicMessageProducer_0_10.java | 15 +- .../client/messaging/address/AddressHelper.java | 11 +- .../main/java/org/apache/qpid/util/FileUtils.java | 2 +- qpid/java/module.xml | 3 + .../queue/MultipleTransactedBatchProducerTest.java | 246 +++++++++++++++++++++ .../destination/AddressBasedDestinationTest.java | 67 +++--- .../qpid/test/unit/message/JMSPropertiesTest.java | 38 ++++ .../apache/qpid/test/utils/QpidBrokerTestCase.java | 36 ++- qpid/java/test-profiles/08StandaloneExcludes | 2 + qpid/java/test-profiles/JavaInVMExcludes | 2 + 13 files changed, 456 insertions(+), 76 deletions(-) create mode 100644 qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MultipleTransactedBatchProducerTest.java (limited to 'qpid/java') diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java index b003152db6..0e3f7b2625 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java @@ -1808,14 +1808,40 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener } + /** + * Used by queue Runners to asynchronously deliver messages to consumers. + * + * A queue Runner is started whenever a state change occurs, e.g when a new + * message arrives on the queue and cannot be immediately delivered to a + * subscription (i.e. asynchronous delivery is required). Unless there are + * SubFlushRunners operating (due to subscriptions unsuspending) which are + * capable of accepting/delivering all messages then these messages would + * otherwise remain on the queue. + * + * processQueue should be running while there are messages on the queue AND + * there are subscriptions that can deliver them. If there are no + * subscriptions capable of delivering the remaining messages on the queue + * then processQueue should stop to prevent spinning. + * + * Since processQueue is runs in a fixed size Executor, it should not run + * indefinitely to prevent starving other tasks of CPU (e.g jobs to process + * incoming messages may not be able to be scheduled in the thread pool + * because all threads are working on clearing down large queues). To solve + * this problem, after an arbitrary number of message deliveries the + * processQueue job stops iterating, resubmits itself to the executor, and + * ends the current instance + * + * @param runner the Runner to schedule + * @throws AMQException + */ private void processQueue(Runnable runner) throws AMQException { long stateChangeCount; long previousStateChangeCount = Long.MIN_VALUE; boolean deliveryIncomplete = true; - int extraLoops = 1; - long iterations = MAX_ASYNC_DELIVERIES; + boolean lastLoop = false; + int iterations = MAX_ASYNC_DELIVERIES; _asynchronousRunner.compareAndSet(runner, null); @@ -1832,12 +1858,14 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener if (previousStateChangeCount != stateChangeCount) { - extraLoops = 1; + //further asynchronous delivery is required since the + //previous loop. keep going if iteration slicing allows. + lastLoop = false; } previousStateChangeCount = stateChangeCount; - deliveryIncomplete = _subscriptionList.size() != 0; - boolean done; + boolean allSubscriptionsDone = true; + boolean subscriptionDone; SubscriptionList.SubscriptionNodeIterator subscriptionIter = _subscriptionList.iterator(); //iterate over the subscribers and try to advance their pointer @@ -1847,30 +1875,25 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener sub.getSendLock(); try { - - done = attemptDelivery(sub); - - if (done) + //attempt delivery. returns true if no further delivery currently possible to this sub + subscriptionDone = attemptDelivery(sub); + if (subscriptionDone) { - if (extraLoops == 0) + //close autoClose subscriptions if we are not currently intent on continuing + if (lastLoop && sub.isAutoClose()) { - deliveryIncomplete = false; - if (sub.isAutoClose()) - { - unregisterSubscription(sub); + unregisterSubscription(sub); - sub.confirmAutoClose(); - } - } - else - { - extraLoops--; + sub.confirmAutoClose(); } } else { + //this subscription can accept additional deliveries, so we must + //keep going after this (if iteration slicing allows it) + allSubscriptionsDone = false; + lastLoop = false; iterations--; - extraLoops = 1; } } finally @@ -1878,10 +1901,34 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener sub.releaseSendLock(); } } + + if(allSubscriptionsDone && lastLoop) + { + //We have done an extra loop already and there are again + //again no further delivery attempts possible, only + //keep going if state change demands it. + deliveryIncomplete = false; + } + else if(allSubscriptionsDone) + { + //All subscriptions reported being done, but we have to do + //an extra loop if the iterations are not exhausted and + //there is still any work to be done + deliveryIncomplete = _subscriptionList.size() != 0; + lastLoop = true; + } + else + { + //some subscriptions can still accept more messages, + //keep going if iteration count allows. + lastLoop = false; + deliveryIncomplete = true; + } + _asynchronousRunner.set(null); } - // If deliveries == 0 then the limitting factor was the time-slicing rather than available messages or credit + // If iterations == 0 then the limiting factor was the time-slicing rather than available messages or credit // therefore we should schedule this runner again (unless someone beats us to it :-) ). if (iterations == 0 && _asynchronousRunner.compareAndSet(null, runner)) { diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java index 75bd50e3a2..54cd709af3 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java @@ -88,8 +88,18 @@ public class ServerConnection extends Connection implements AMQConnectionModel, _onOpenTask.run(); } _actor.message(ConnectionMessages.OPEN(getClientId(), "0-10", true, true)); + + getVirtualHost().getConnectionRegistry().registerConnection(this); } - + + if (state == State.CLOSE_RCVD || state == State.CLOSED || state == State.CLOSING) + { + if(_virtualHost != null) + { + _virtualHost.getConnectionRegistry().deregisterConnection(this); + } + } + if (state == State.CLOSED) { logClosed(); @@ -126,7 +136,6 @@ public class ServerConnection extends Connection implements AMQConnectionModel, public void setVirtualHost(VirtualHost virtualHost) { _virtualHost = virtualHost; - _virtualHost.getConnectionRegistry().registerConnection(this); initialiseStatistics(); } @@ -253,7 +262,6 @@ public class ServerConnection extends Connection implements AMQConnectionModel, // Ignore } close(replyCode, message); - getVirtualHost().getConnectionRegistry().deregisterConnection(this); } @Override diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java index 5fc8d43e94..d518d7b2c4 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java @@ -3461,4 +3461,9 @@ public abstract class AMQSession + + @@ -346,6 +348,7 @@ + diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MultipleTransactedBatchProducerTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MultipleTransactedBatchProducerTest.java new file mode 100644 index 0000000000..460270e188 --- /dev/null +++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MultipleTransactedBatchProducerTest.java @@ -0,0 +1,246 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.server.queue; + +import java.util.Random; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import javax.jms.Connection; +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.MessageConsumer; +import javax.jms.MessageListener; +import javax.jms.Session; + +import org.apache.log4j.Logger; +import org.apache.qpid.test.utils.QpidBrokerTestCase; + +public class MultipleTransactedBatchProducerTest extends QpidBrokerTestCase +{ + private static final Logger _logger = Logger.getLogger(MultipleTransactedBatchProducerTest.class); + + private static final int MESSAGE_COUNT = 1000; + private static final int BATCH_SIZE = 50; + private static final int NUM_PRODUCERS = 2; + private static final int NUM_CONSUMERS = 3; + private static final Random RANDOM = new Random(); + + private CountDownLatch _receivedLatch; + private String _queueName; + + private volatile String _failMsg; + + public void setUp() throws Exception + { + //debug level logging often makes this test pass artificially, turn the level down to info. + setSystemProperty("amqj.server.logging.level", "INFO"); + _receivedLatch = new CountDownLatch(MESSAGE_COUNT * NUM_PRODUCERS); + setConfigurationProperty("management.enabled", "true"); + super.setUp(); + _queueName = getTestQueueName(); + _failMsg = null; + } + + /** + * When there are multiple producers submitting batches of messages to a given + * queue using transacted sessions, it is highly probable that concurrent + * enqueue() activity will occur and attempt delivery of their message to the + * same subscription. In this scenario it is likely that one of the attempts + * will succeed and the other will result in use of the deliverAsync() method + * to start a queue Runner and ensure delivery of the message. + * + * A defect within the processQueue() method used by the Runner would mean that + * delivery of these messages may not occur, should the Runner stop before all + * messages have been processed. Such a defect was discovered and found to be + * most visible when Selectors are used such that one and only one subscription + * can/will accept any given message, but multiple subscriptions are present, + * and one of the earlier subscriptions receives more messages than the others. + * + * This test is to validate that the processQueue() method is able to correctly + * deliver all of the messages present for asynchronous delivery to subscriptions, + * by utilising multiple batch transacted producers to create the scenario and + * ensure all messages are received by a consumer. + */ + public void testMultipleBatchedProducersWithMultipleConsumersUsingSelectors() throws Exception + { + String selector1 = ("(\"" + _queueName +"\" % " + NUM_CONSUMERS + ") = 0"); + String selector2 = ("(\"" + _queueName +"\" % " + NUM_CONSUMERS + ") = 1"); + String selector3 = ("(\"" + _queueName +"\" % " + NUM_CONSUMERS + ") = 2"); + + //create consumers + Connection conn1 = getConnection(); + conn1.setExceptionListener(new ExceptionHandler("conn1")); + Session sess1 = conn1.createSession(true, Session.SESSION_TRANSACTED); + MessageConsumer cons1 = sess1.createConsumer(sess1.createQueue(_queueName), selector1); + cons1.setMessageListener(new Cons(sess1,"consumer1")); + + Connection conn2 = getConnection(); + conn2.setExceptionListener(new ExceptionHandler("conn2")); + Session sess2 = conn2.createSession(true, Session.SESSION_TRANSACTED); + MessageConsumer cons2 = sess2.createConsumer(sess2.createQueue(_queueName), selector2); + cons2.setMessageListener(new Cons(sess2,"consumer2")); + + Connection conn3 = getConnection(); + conn3.setExceptionListener(new ExceptionHandler("conn3")); + Session sess3 = conn3.createSession(true, Session.SESSION_TRANSACTED); + MessageConsumer cons3 = sess3.createConsumer(sess3.createQueue(_queueName), selector3); + cons3.setMessageListener(new Cons(sess3,"consumer3")); + + conn1.start(); + conn2.start(); + conn3.start(); + + //create producers + Connection connA = getConnection(); + connA.setExceptionListener(new ExceptionHandler("connA")); + Connection connB = getConnection(); + connB.setExceptionListener(new ExceptionHandler("connB")); + Thread producer1 = new Thread(new ProducerThread(connA, _queueName, "producer1")); + Thread producer2 = new Thread(new ProducerThread(connB, _queueName, "producer2")); + + producer1.start(); + Thread.sleep(10); + producer2.start(); + + //await delivery of the messages + boolean result = _receivedLatch.await(75, TimeUnit.SECONDS); + + assertNull("Test failed because: " + String.valueOf(_failMsg), _failMsg); + assertTrue("Some of the messages were not all recieved in the given timeframe, remaining count was: "+_receivedLatch.getCount(), + result); + + } + + @Override + public Message createNextMessage(Session session, int msgCount) throws JMSException + { + Message message = super.createNextMessage(session,msgCount); + + //bias at least 50% of the messages to the first consumers selector because + //the issue presents itself primarily when an earlier subscription completes + //delivery after the later subscriptions + int val; + if (msgCount % 2 == 0) + { + val = 0; + } + else + { + val = RANDOM.nextInt(Integer.MAX_VALUE); + } + + message.setIntProperty(_queueName, val); + + return message; + } + + private class Cons implements MessageListener + { + private Session _sess; + private String _desc; + + public Cons(Session sess, String desc) + { + _sess = sess; + _desc = desc; + } + + public void onMessage(Message message) + { + _receivedLatch.countDown(); + int msgCount = 0; + int msgID = 0; + try + { + msgCount = message.getIntProperty(INDEX); + msgID = message.getIntProperty(_queueName); + } + catch (JMSException e) + { + _logger.error(_desc + " received exception: " + e.getMessage(), e); + failAsyncTest(e.getMessage()); + } + + _logger.info("Consumer received message:"+ msgCount + " with ID: " + msgID); + + try + { + _sess.commit(); + } + catch (JMSException e) + { + _logger.error(_desc + " received exception: " + e.getMessage(), e); + failAsyncTest(e.getMessage()); + } + } + } + + private class ProducerThread implements Runnable + { + private Connection _conn; + private String _dest; + private String _desc; + + public ProducerThread(Connection conn, String dest, String desc) + { + _conn = conn; + _dest = dest; + _desc = desc; + } + + public void run() + { + try + { + Session session = _conn.createSession(true, Session.SESSION_TRANSACTED); + sendMessage(session, session.createQueue(_dest), MESSAGE_COUNT, BATCH_SIZE); + } + catch (Exception e) + { + _logger.error(_desc + " received exception: " + e.getMessage(), e); + failAsyncTest(e.getMessage()); + } + } + } + + private class ExceptionHandler implements javax.jms.ExceptionListener + { + private String _desc; + + public ExceptionHandler(String description) + { + _desc = description; + } + + public void onException(JMSException e) + { + _logger.error(_desc + " received exception: " + e.getMessage(), e); + failAsyncTest(e.getMessage()); + } + } + + private void failAsyncTest(String msg) + { + _logger.error("Failing test because: " + msg); + _failMsg = msg; + } +} \ No newline at end of file diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java index 30dc30cd81..a737781e30 100644 --- a/qpid/java/systests/src/main/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java +++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java @@ -473,39 +473,6 @@ public class AddressBasedDestinationTest extends QpidBrokerTestCase } - /** - * Test goal: Verifies that and address based destination can be used successfully - * as a reply to. - */ - public void testAddressBasedReplyTo() throws Exception - { - Session jmsSession = _connection.createSession(false,Session.AUTO_ACKNOWLEDGE); - - String addr = "ADDR:amq.direct/x512; {create: receiver, " + - "link : {name : 'MY.RESP.QUEUE', " + - "x-declare : { auto-delete: true, exclusive: true, " + - "arguments : {'qpid.max_size': 1000, 'qpid.policy_type': ring }} } }"; - - Destination replyTo = new AMQAnyDestination(addr); - Destination dest =new AMQAnyDestination("ADDR:amq.direct/Hello"); - - MessageConsumer cons = jmsSession.createConsumer(dest); - MessageProducer prod = jmsSession.createProducer(dest); - Message m = jmsSession.createTextMessage("Hello"); - m.setJMSReplyTo(replyTo); - prod.send(m); - - Message msg = cons.receive(1000); - assertNotNull("consumer should have received the message",msg); - - MessageConsumer replyToCons = jmsSession.createConsumer(replyTo); - MessageProducer replyToProd = jmsSession.createProducer(msg.getJMSReplyTo()); - replyToProd.send(jmsSession.createTextMessage("reply")); - - Message replyToMsg = replyToCons.receive(1000); - assertNotNull("The reply to consumer should have got the message",replyToMsg); - } - /** * Test goal: Verifies that session.createQueue method * works as expected both with the new and old addressing scheme. @@ -1020,4 +987,38 @@ public class AddressBasedDestinationTest extends QpidBrokerTestCase prod.close(); cons.close(); } + + public void testReplyToWithNamelessExchange() throws Exception + { + System.setProperty("qpid.declare_exchanges","false"); + replyToTest("ADDR:my-queue;{create: always}"); + System.setProperty("qpid.declare_exchanges","true"); + } + + public void testReplyToWithCustomExchange() throws Exception + { + replyToTest("ADDR:hello;{create:always,node:{type:topic}}"); + } + + private void replyToTest(String replyTo) throws Exception + { + Session session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Destination replyToDest = AMQDestination.createDestination(replyTo); + MessageConsumer replyToCons = session.createConsumer(replyToDest); + + Destination dest = session.createQueue("amq.direct/test"); + + MessageConsumer cons = session.createConsumer(dest); + MessageProducer prod = session.createProducer(dest); + Message m = session.createTextMessage("test"); + m.setJMSReplyTo(replyToDest); + prod.send(m); + + Message msg = cons.receive(); + MessageProducer prodR = session.createProducer(msg.getJMSReplyTo()); + prodR.send(session.createTextMessage("x")); + + Message m1 = replyToCons.receive(); + assertNotNull("The reply to consumer should have received the messsage",m1); + } } diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/message/JMSPropertiesTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/message/JMSPropertiesTest.java index 830421a01f..8caeaa55c0 100644 --- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/message/JMSPropertiesTest.java +++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/message/JMSPropertiesTest.java @@ -31,6 +31,7 @@ import org.apache.qpid.test.utils.QpidBrokerTestCase; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.jms.Connection; import javax.jms.Destination; import javax.jms.Message; import javax.jms.MessageConsumer; @@ -39,7 +40,11 @@ import javax.jms.MessageProducer; import javax.jms.ObjectMessage; import javax.jms.Queue; import javax.jms.Session; +import javax.jms.Topic; + import java.util.Enumeration; +import java.util.HashMap; +import java.util.Map; /** * @author Apache Software Foundation @@ -163,4 +168,37 @@ public class JMSPropertiesTest extends QpidBrokerTestCase con.close(); } + /** + * Test Goal : test if the message properties can be retrieved properly with out an error + * and also test if unsupported properties are filtered out. See QPID-2930. + */ + public void testGetPropertyNames() throws Exception + { + Connection con = getConnection("guest", "guest"); + Session ssn = (AMQSession) con.createSession(false, Session.CLIENT_ACKNOWLEDGE); + con.start(); + + Topic topic = ssn.createTopic("test"); + MessageConsumer consumer = ssn.createConsumer(topic); + MessageProducer prod = ssn.createProducer(topic); + Message m = ssn.createMessage(); + m.setObjectProperty("x-amqp-0-10.routing-key", "routing-key".getBytes()); + m.setObjectProperty("routing-key", "routing-key"); + prod.send(m); + + Message msg = consumer.receive(1000); + assertNotNull(msg); + + Enumeration enu = msg.getPropertyNames(); + Map map = new HashMap(); + while (enu.hasMoreElements()) + { + String name = enu.nextElement(); + String value = msg.getStringProperty(name); + map.put(name, value); + } + + assertFalse("Property 'x-amqp-0-10.routing-key' should have been filtered out",map.containsKey("x-amqp-0-10.routing-key")); + assertTrue("Property routing-key should be present",map.containsKey("routing-key")); + } } diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java index ae38a75e7a..d7da5d8a78 100644 --- a/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java +++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java @@ -67,6 +67,7 @@ import org.apache.qpid.server.registry.ApplicationRegistry; import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry; import org.apache.qpid.server.store.DerbyMessageStore; import org.apache.qpid.url.URLSyntaxException; +import org.apache.qpid.util.FileUtils; import org.apache.qpid.util.LogMonitor; /** @@ -109,6 +110,7 @@ public class QpidBrokerTestCase extends QpidTestCase private static final String BROKER = "broker"; private static final String BROKER_CLEAN = "broker.clean"; private static final String BROKER_CLEAN_BETWEEN_TESTS = "broker.clean.between.tests"; + private static final String BROKER_EXISTING_QPID_WORK = "broker.existing.qpid.work"; private static final String BROKER_VERSION = "broker.version"; protected static final String BROKER_READY = "broker.ready"; private static final String BROKER_STOPPED = "broker.stopped"; @@ -283,6 +285,19 @@ public class QpidBrokerTestCase extends QpidTestCase fail("Unable to test without config file:" + _configFile); } + if(_brokerCleanBetweenTests) + { + cleanBroker(); + + String existingQpidWorkPath = System.getProperty(BROKER_EXISTING_QPID_WORK); + if(existingQpidWorkPath != null && !existingQpidWorkPath.equals("")) + { + File existing = new File(existingQpidWorkPath); + File qpidWork = new File(getQpidWork(_broker, getPort())); + FileUtils.copyRecursive(existing, qpidWork); + } + } + startBroker(); } @@ -490,7 +505,7 @@ public class QpidBrokerTestCase extends QpidTestCase // DON'T change PNAME, qpid.stop needs this value. env.put("QPID_PNAME", "-DPNAME=QPBRKR -DTNAME=\"" + _testName + "\""); // Add the port to QPID_WORK to ensure unique working dirs for multi broker tests - env.put("QPID_WORK", System.getProperty("QPID_WORK")+ "/" + port); + env.put("QPID_WORK", getQpidWork(_broker, port)); // Use the environment variable to set amqj.logging.level for the broker @@ -578,6 +593,20 @@ public class QpidBrokerTestCase extends QpidTestCase _brokers.put(port, process); } + private String getQpidWork(String broker, int port) + { + if (broker.equals(VM)) + { + return System.getProperty("QPID_WORK"); + } + else if (!broker.equals(EXTERNAL)) + { + return System.getProperty("QPID_WORK")+ "/" + port; + } + + return System.getProperty("QPID_WORK"); + } + public String getTestConfigFile() { String path = _output == null ? System.getProperty("java.io.tmpdir") : _output; @@ -1190,7 +1219,8 @@ public class QpidBrokerTestCase extends QpidTestCase MessageProducer producer = session.createProducer(destination); - for (int i = offset; i < (count + offset); i++) + int i = offset; + for (; i < (count + offset); i++) { Message next = createNextMessage(session, i); @@ -1213,7 +1243,7 @@ public class QpidBrokerTestCase extends QpidTestCase // we have no batchSize or // our count is not divible by batchSize. if (session.getTransacted() && - ( batchSize == 0 || count % batchSize != 0)) + ( batchSize == 0 || (i-1) % batchSize != 0)) { session.commit(); } diff --git a/qpid/java/test-profiles/08StandaloneExcludes b/qpid/java/test-profiles/08StandaloneExcludes index b482a14c6d..43eb1f8ee5 100644 --- a/qpid/java/test-profiles/08StandaloneExcludes +++ b/qpid/java/test-profiles/08StandaloneExcludes @@ -37,3 +37,5 @@ org.apache.qpid.test.unit.message.UTF8Test#* org.apache.qpid.client.MessageListenerTest#testSynchronousReceiveNoWait org.apache.qpid.test.unit.client.connection.ConnectionTest#testUnsupportedSASLMechanism + +org.apache.qpid.test.unit.message.JMSPropertiesTest#testGetPropertyNames diff --git a/qpid/java/test-profiles/JavaInVMExcludes b/qpid/java/test-profiles/JavaInVMExcludes index 7e7da4302e..7960b28d81 100644 --- a/qpid/java/test-profiles/JavaInVMExcludes +++ b/qpid/java/test-profiles/JavaInVMExcludes @@ -41,3 +41,5 @@ org.apache.qpid.test.unit.ack.RecoverTest#testOderingWithSyncConsumer //The VM broker does not export the logging management JMX MBean org.apache.qpid.server.security.acl.ExternalAdminACLTest#* + +org.apache.qpid.test.unit.message.JMSPropertiesTest#testGetPropertyNames -- cgit v1.2.1