Author: ritchiem Date: 2008-05-20 06:47:17 -0400 (Tue, 20 May 2008) New Revision: 2065 Added: store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/ store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/ store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/ store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/StoreContextRaceConditionTest.java Removed: store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/ store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/ store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/StoreContextRaceConditionTest.java Modified: store/branches/java/M2.1.x/ store/branches/java/M2.1.x/java/bdbstore/pom.xml Log: Merged revisions 1922-2064 via svnmerge from https://svn.jboss.org/repos/rhmessaging/store/branches/java/M2.x ........ r1923 | ritchiem | 2008-04-16 17:14:16 +0100 (Wed, 16 Apr 2008) | 1 line Added test for RaceCondtion in present between BDBStore and Qpid Housekeeping thread. ........ Property changes on: store/branches/java/M2.1.x ___________________________________________________________________ Name: svnmerge-integrated - /store/branches/java/M2.x:1-1921 + /store/branches/java/M2.x:1-2064 Modified: store/branches/java/M2.1.x/java/bdbstore/pom.xml =================================================================== --- store/branches/java/M2.1.x/java/bdbstore/pom.xml 2008-05-20 10:45:51 UTC (rev 2064) +++ store/branches/java/M2.1.x/java/bdbstore/pom.xml 2008-05-20 10:47:17 UTC (rev 2065) @@ -32,6 +32,10 @@ <version>1.0-incubating-M2.1.x-SNAPSHOT</version> </parent> + <properties> + <topDirectoryLocation>..</topDirectoryLocation> + </properties> + <!-- Local repository for the BerkelyDB-je so we don't have to use the installer script --> <repositories> <repository> @@ -78,6 +82,25 @@ <build> <plugins> + + + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-surefire-plugin</artifactId> + <configuration> + <systemProperties> + <property> + <name>QPID_HOME</name> + <value>${basedir}/${topDirectoryLocation}/broker</value> + </property> + </systemProperties> + <excludes> + <exclude>**/StoreContextRaceConditionTest.class</exclude> + </excludes> + </configuration> + </plugin> + + <!-- Produce an all dependencies jar, to be used for running the hot backup utility as a standalone tool. --> <plugin> <groupId>org.apache.maven.plugins</groupId> Copied: store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp (from rev 1923, store/branches/java/M2.x/java/bdbstore/src/test/java/org/etp) Copied: store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid (from rev 1923, store/branches/java/M2.x/java/bdbstore/src/test/java/org/etp/qpid) Copied: store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server (from rev 1923, store/branches/java/M2.x/java/bdbstore/src/test/java/org/etp/qpid/server) Deleted: store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/StoreContextRaceConditionTest.java =================================================================== --- store/branches/java/M2.x/java/bdbstore/src/test/java/org/etp/qpid/server/StoreContextRaceConditionTest.java 2008-04-16 16:14:16 UTC (rev 1923) +++ store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/StoreContextRaceConditionTest.java 2008-05-20 10:47:17 UTC (rev 2065) @@ -1,271 +0,0 @@ -/* - * - * 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.etp.qpid.server; - -import org.apache.commons.configuration.Configuration; -import org.apache.log4j.Level; -import org.apache.log4j.Logger; -import org.apache.log4j.BasicConfigurator; -import org.apache.qpid.server.registry.ApplicationRegistry; -import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry; -import org.apache.qpid.server.virtualhost.VirtualHostRegistry; -import org.apache.qpid.test.VMTestCase; - -import javax.jms.Connection; -import javax.jms.ConnectionFactory; -import javax.jms.JMSException; -import javax.jms.MessageConsumer; -import javax.jms.MessageProducer; -import javax.jms.Queue; -import javax.jms.Session; -import javax.jms.TextMessage; -import javax.naming.NamingException; -import java.io.File; - -public class StoreContextRaceConditionTest extends VMTestCase -{ - private static final Logger _logger = Logger.getLogger(StoreContextRaceConditionTest.class); - private long _TimeToLive = 0L; - private static long SECOND = 1000L; - - private static final String LOGGING_KEY = "amqj.logging.level"; - - private String _loggingOriginal; - - public void setUp() throws Exception - { - //Disable the logging - _loggingOriginal = System.getProperty(LOGGING_KEY); - System.setProperty(LOGGING_KEY, "WARN"); - - // Initialise ACLs. - final String qpidHome = System.getProperty("QPID_HOME"); - - final File defaultaclConfigFile = new File(qpidHome, "etc/persistent_config.xml"); - - if (!defaultaclConfigFile.exists()) - { - System.err.println("Configuration file not found:" + defaultaclConfigFile); - fail("Configuration file not found:" + defaultaclConfigFile); - } - - if (System.getProperty("QPID_HOME") == null) - { - fail("QPID_HOME not set"); - } - - Configuration configuration = ConfigurationFileApplicationRegistry.config(defaultaclConfigFile); - - //Reset the housekeeping threads to run every second. - configuration.setProperty("virtualhosts.virtualhost.test.housekeeping.expiredMessageCheckPeriod", 10 * SECOND); - configuration.setProperty("management.enabled", "false"); - - _logger.info("Reset Housekeeping to :" + configuration.getLong("virtualhosts.virtualhost.test.housekeeping.expiredMessageCheckPeriod")); - - ConfigurationFileApplicationRegistry config = new ConfigurationFileApplicationRegistry(configuration); - - ApplicationRegistry.initialise(config, 1); - - //Remove the Vhosts we are not using to free up CPU from extra housekeeping threads. - VirtualHostRegistry vHostRegistry = ApplicationRegistry.getInstance().getVirtualHostRegistry(); - vHostRegistry.getVirtualHost("localhost").close(); - vHostRegistry.getVirtualHost("development").close(); - - //Create the Broker - super.setUp(); - - _queue = (Queue) _context.lookup("queue"); - } - - public void tearDown() - { - System.setProperty(LOGGING_KEY, _loggingOriginal); - } - - protected static final String MESSAGE_ID_PROPERTY = "MessageIDProperty"; - - protected Queue _queue; - - protected void sendMessages(int num) throws JMSException - { - Connection producerConnection = null; - try - { - producerConnection = ((ConnectionFactory) _context.lookup("connection")).createConnection(); - } - catch (NamingException e) - { - fail("Unable to lookup connection in JNDI."); - } - - sendMessages(producerConnection, num); - } - - protected void sendMessages(Connection producerConnection, int num) throws JMSException - { - Session producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); - - //Ensure _queue is created - producerSession.createConsumer(_queue).close(); - - MessageProducer producer = producerSession.createProducer(_queue); - - producer.setTimeToLive(_TimeToLive); - producer.setDisableMessageTimestamp(false); - - for (int messsageID = 0; messsageID < num; messsageID++) - { - TextMessage textMsg = producerSession.createTextMessage("Message " + messsageID); - textMsg.setIntProperty(MESSAGE_ID_PROPERTY, messsageID); - producer.send(textMsg); - } - - producerConnection.close(); - } - - public void test() throws InterruptedException, NamingException, JMSException - { - Runnable test = new Runnable() - { - public void run() - { - - //Create Consumer - Connection connection = null; - - Session session = null; - try - { - try - { - connection = ((ConnectionFactory) _context.lookup("connection")).createConnection(); - } - catch (NamingException e) - { - fail("Unable to lookup connection in JNDI."); - } - session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); - } - catch (JMSException e) - { - return; - } - - try - { - int run = 1; - while (true) - { - try - { - //Stop the connection to prevent flow - connection.stop(); - //Create Consumer to receive msgs - MessageConsumer consumer = session.createConsumer(_queue); - - //Send one message to hold up the Async Delivery from purging - _logger.info("***** CREATED Consumer"); - _TimeToLive = 0L; - sendMessages(1); - _logger.info("***** SENT msg 1"); - //Send 1000 msgs that will time out - _TimeToLive = 1000L; - sendMessages(50); - _logger.info("***** SENT TTL msgs"); - - //Timeout Messages - Note that we - Thread.sleep(1000); - _logger.info("***** SLEEP"); - - //Allw the messages to flow to us - connection.start(); - _logger.info("***** START Consumer"); - //*** Starts Async process - - //Remove the first message so that the async will occcur and start purging. - consumer.receive(1000); - _logger.info("***** RECEIVE Consumer"); - - sendMessages(50); - _logger.info("***** SENT TTL msgs"); - - //Close the consumer freeing the QHK thread to doing work - consumer.close(); - _logger.info("***** CLOSE Consumer"); - //** Allows QueueHouskeeping to run. - sendMessages(50); - _logger.info("***** SENT TTL msgs"); - - System.err.println("********************************* Running test again (" + run + - ")in attempt to cause race condition."); - run++; - } - catch (JMSException e) - { - - } - catch (InterruptedException e) - { - } - } - } - finally - { - try - { - connection.close(); - } - catch (JMSException e) - { - e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. - } - _logger.info("***** Test Done"); - } - } - }; - - int MAX_THREADS = 1; - - Thread[] threads = new Thread[MAX_THREADS]; - - for (int concurentClients = 0; concurentClients < MAX_THREADS; concurentClients++) - { - threads[concurentClients] = new Thread(test); - threads[concurentClients].start(); - } - - for (int concurentClients = 0; concurentClients < MAX_THREADS; concurentClients++) - { - threads[concurentClients].join(); - } - } - - public static void main(String[] args) throws Exception, InterruptedException - { - StoreContextRaceConditionTest scrc = new StoreContextRaceConditionTest(); - - scrc.setUp(); - scrc.test(); -// scrc.tearDown(); - } - -} Copied: store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/StoreContextRaceConditionTest.java (from rev 1923, store/branches/java/M2.x/java/bdbstore/src/test/java/org/etp/qpid/server/StoreContextRaceConditionTest.java) =================================================================== --- store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/StoreContextRaceConditionTest.java (rev 0) +++ store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/StoreContextRaceConditionTest.java 2008-05-20 10:47:17 UTC (rev 2065) @@ -0,0 +1,271 @@ +/* + * + * 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.etp.qpid.server; + +import org.apache.commons.configuration.Configuration; +import org.apache.log4j.Level; +import org.apache.log4j.Logger; +import org.apache.log4j.BasicConfigurator; +import org.apache.qpid.server.registry.ApplicationRegistry; +import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry; +import org.apache.qpid.server.virtualhost.VirtualHostRegistry; +import org.apache.qpid.test.VMTestCase; + +import javax.jms.Connection; +import javax.jms.ConnectionFactory; +import javax.jms.JMSException; +import javax.jms.MessageConsumer; +import javax.jms.MessageProducer; +import javax.jms.Queue; +import javax.jms.Session; +import javax.jms.TextMessage; +import javax.naming.NamingException; +import java.io.File; + +public class StoreContextRaceConditionTest extends VMTestCase +{ + private static final Logger _logger = Logger.getLogger(StoreContextRaceConditionTest.class); + private long _TimeToLive = 0L; + private static long SECOND = 1000L; + + private static final String LOGGING_KEY = "amqj.logging.level"; + + private String _loggingOriginal; + + public void setUp() throws Exception + { + //Disable the logging + _loggingOriginal = System.getProperty(LOGGING_KEY); + System.setProperty(LOGGING_KEY, "WARN"); + + // Initialise ACLs. + final String qpidHome = System.getProperty("QPID_HOME"); + + final File defaultaclConfigFile = new File(qpidHome, "etc/persistent_config.xml"); + + if (!defaultaclConfigFile.exists()) + { + System.err.println("Configuration file not found:" + defaultaclConfigFile); + fail("Configuration file not found:" + defaultaclConfigFile); + } + + if (System.getProperty("QPID_HOME") == null) + { + fail("QPID_HOME not set"); + } + + Configuration configuration = ConfigurationFileApplicationRegistry.config(defaultaclConfigFile); + + //Reset the housekeeping threads to run every second. + configuration.setProperty("virtualhosts.virtualhost.test.housekeeping.expiredMessageCheckPeriod", 10 * SECOND); + configuration.setProperty("management.enabled", "false"); + + _logger.info("Reset Housekeeping to :" + configuration.getLong("virtualhosts.virtualhost.test.housekeeping.expiredMessageCheckPeriod")); + + ConfigurationFileApplicationRegistry config = new ConfigurationFileApplicationRegistry(configuration); + + ApplicationRegistry.initialise(config, 1); + + //Remove the Vhosts we are not using to free up CPU from extra housekeeping threads. + VirtualHostRegistry vHostRegistry = ApplicationRegistry.getInstance().getVirtualHostRegistry(); + vHostRegistry.getVirtualHost("localhost").close(); + vHostRegistry.getVirtualHost("development").close(); + + //Create the Broker + super.setUp(); + + _queue = (Queue) _context.lookup("queue"); + } + + public void tearDown() + { + System.setProperty(LOGGING_KEY, _loggingOriginal); + } + + protected static final String MESSAGE_ID_PROPERTY = "MessageIDProperty"; + + protected Queue _queue; + + protected void sendMessages(int num) throws JMSException + { + Connection producerConnection = null; + try + { + producerConnection = ((ConnectionFactory) _context.lookup("connection")).createConnection(); + } + catch (NamingException e) + { + fail("Unable to lookup connection in JNDI."); + } + + sendMessages(producerConnection, num); + } + + protected void sendMessages(Connection producerConnection, int num) throws JMSException + { + Session producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); + + //Ensure _queue is created + producerSession.createConsumer(_queue).close(); + + MessageProducer producer = producerSession.createProducer(_queue); + + producer.setTimeToLive(_TimeToLive); + producer.setDisableMessageTimestamp(false); + + for (int messsageID = 0; messsageID < num; messsageID++) + { + TextMessage textMsg = producerSession.createTextMessage("Message " + messsageID); + textMsg.setIntProperty(MESSAGE_ID_PROPERTY, messsageID); + producer.send(textMsg); + } + + producerConnection.close(); + } + + public void test() throws InterruptedException, NamingException, JMSException + { + Runnable test = new Runnable() + { + public void run() + { + + //Create Consumer + Connection connection = null; + + Session session = null; + try + { + try + { + connection = ((ConnectionFactory) _context.lookup("connection")).createConnection(); + } + catch (NamingException e) + { + fail("Unable to lookup connection in JNDI."); + } + session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + } + catch (JMSException e) + { + return; + } + + try + { + int run = 1; + while (true) + { + try + { + //Stop the connection to prevent flow + connection.stop(); + //Create Consumer to receive msgs + MessageConsumer consumer = session.createConsumer(_queue); + + //Send one message to hold up the Async Delivery from purging + _logger.info("***** CREATED Consumer"); + _TimeToLive = 0L; + sendMessages(1); + _logger.info("***** SENT msg 1"); + //Send 1000 msgs that will time out + _TimeToLive = 1000L; + sendMessages(50); + _logger.info("***** SENT TTL msgs"); + + //Timeout Messages - Note that we + Thread.sleep(1000); + _logger.info("***** SLEEP"); + + //Allw the messages to flow to us + connection.start(); + _logger.info("***** START Consumer"); + //*** Starts Async process + + //Remove the first message so that the async will occcur and start purging. + consumer.receive(1000); + _logger.info("***** RECEIVE Consumer"); + + sendMessages(50); + _logger.info("***** SENT TTL msgs"); + + //Close the consumer freeing the QHK thread to doing work + consumer.close(); + _logger.info("***** CLOSE Consumer"); + //** Allows QueueHouskeeping to run. + sendMessages(50); + _logger.info("***** SENT TTL msgs"); + + System.err.println("********************************* Running test again (" + run + + ")in attempt to cause race condition."); + run++; + } + catch (JMSException e) + { + + } + catch (InterruptedException e) + { + } + } + } + finally + { + try + { + connection.close(); + } + catch (JMSException e) + { + e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. + } + _logger.info("***** Test Done"); + } + } + }; + + int MAX_THREADS = 1; + + Thread[] threads = new Thread[MAX_THREADS]; + + for (int concurentClients = 0; concurentClients < MAX_THREADS; concurentClients++) + { + threads[concurentClients] = new Thread(test); + threads[concurentClients].start(); + } + + for (int concurentClients = 0; concurentClients < MAX_THREADS; concurentClients++) + { + threads[concurentClients].join(); + } + } + + public static void main(String[] args) throws Exception, InterruptedException + { + StoreContextRaceConditionTest scrc = new StoreContextRaceConditionTest(); + + scrc.setUp(); + scrc.test(); +// scrc.tearDown(); + } + +}