[hornetq-commits] JBoss hornetq SVN: r11091 - in branches/Branch_2_2_EAP_cluster_clean2: tests/src/org/hornetq/tests/integration/client and 3 other directories.

Author: clebert.suconic(a)jboss.com Date: 2011-08-01 15:39:43 -0400 (Mon, 01 Aug 2011) New Revision: 11091 Modified: branches/Branch_2_2_EAP_cluster_clean2/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/client/JMSPagingFileDeleteTest.java branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/distribution/NettyFileStorageSymmetricClusterWithDiscoveryTest.java branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/distribution/NettySymmetricClusterWithBackupTest.java branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/distribution/OneWayChainClusterTest.java branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTestBase.java branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/failover/NettyAsynchronousReattachTest.java branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/util/UnitTestCase.java Log: log and debug for a test Modified: branches/Branch_2_2_EAP_cluster_clean2/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java =================================================================== --- branches/Branch_2_2_EAP_cluster_clean2/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2011-08-01 16:25:50 UTC (rev 11090) +++ branches/Branch_2_2_EAP_cluster_clean2/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2011-08-01 19:39:43 UTC (rev 11091) @@ -1087,13 +1087,6 @@ { DelegatingBufferHandler handler = new DelegatingBufferHandler(); - if (log.isDebugEnabled()) - { - log.debug("Trying to connect with connector = " + connectorFactory + - ", parameters = " + - connectorConfig.getParams()); - } - connector = connectorFactory.createConnector(connectorConfig.getParams(), handler, this, @@ -1101,6 +1094,15 @@ threadPool, scheduledThreadPool); + if (log.isDebugEnabled()) + { + log.debug("Trying to connect with connector = " + connectorFactory + + ", parameters = " + + connectorConfig.getParams() + " connector = " + connector); + } + + + if (connector != null) { connector.start(); Modified: branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/client/JMSPagingFileDeleteTest.java =================================================================== --- branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/client/JMSPagingFileDeleteTest.java 2011-08-01 16:25:50 UTC (rev 11090) +++ branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/client/JMSPagingFileDeleteTest.java 2011-08-01 19:39:43 UTC (rev 11091) @@ -15,12 +15,15 @@ import javax.jms.BytesMessage; import javax.jms.Connection; +import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageConsumer; import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.Topic; +import junit.framework.Assert; + import org.hornetq.api.core.SimpleString; import org.hornetq.core.logging.Logger; import org.hornetq.core.paging.PagingStore; @@ -30,19 +33,9 @@ public class JMSPagingFileDeleteTest extends JMSTestBase { static Logger log = Logger.getLogger(JMSPagingFileDeleteTest.class); - + Topic topic1; - Connection connection; - - Session session; - - MessageConsumer subscriber1; - - MessageConsumer subscriber2; - - PagingStore pagingStore; - private static final int MESSAGE_SIZE = 1024; private static final int PAGE_SIZE = 10 * 1024; @@ -80,113 +73,195 @@ topic1 = null; super.tearDown(); } - + public void testTopics() throws Exception { - connection = null; + Connection connection = null; try { connection = cf.createConnection(); connection.setClientID("cid"); - session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); MessageProducer producer = session.createProducer(topic1); - subscriber1 = session.createDurableSubscriber(topic1, "subscriber-1"); - subscriber2 = session.createDurableSubscriber(topic1, "subscriber-2"); + MessageConsumer subscriber1 = session.createDurableSubscriber(topic1, "subscriber-1"); + MessageConsumer subscriber2 = session.createDurableSubscriber(topic1, "subscriber-2"); - // -----------------(Step1) Publish Messages to make Paging Files. -------------------- - System.out.println("---------- Send messages. ----------"); - BytesMessage bytesMessage = session.createBytesMessage(); - bytesMessage.writeBytes(new byte[JMSPagingFileDeleteTest.MESSAGE_SIZE]); - for (int i = 0; i < JMSPagingFileDeleteTest.MESSAGE_NUM; i++) - { - producer.send(bytesMessage); - } - System.out.println("Sent " + JMSPagingFileDeleteTest.MESSAGE_NUM + " messages."); + int numMessages = sendMessages(createMessage(session), producer); - pagingStore = server.getPagingManager().getPageStore(new SimpleString("jms.topic.topic1")); - printPageStoreInfo(pagingStore); + printPageStoreInfo(); - assertTrue(pagingStore.isPaging()); + Assert.assertTrue(getPagingStore().isPaging()); connection.start(); // -----------------(Step2) Restart the server. -------------------------------------- - stopAndStartServer(); // If try this test without restarting server, please comment out this line; + // If try this test without restarting server, please comment out this section; + close(connection); + stopAndStartServer(); + connection = cf.createConnection(); + connection.setClientID("cid"); + session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + subscriber1 = session.createDurableSubscriber(topic1, "subscriber-1"); + subscriber2 = session.createDurableSubscriber(topic1, "subscriber-2"); + connection.start(); + // -----------------(Step3) Subscribe to all the messages from the topic.-------------- System.out.println("---------- Receive all messages. ----------"); - for (int i = 0; i < JMSPagingFileDeleteTest.MESSAGE_NUM; i++) + for (int i = 0; i < numMessages; i++) { - Message message1 = subscriber1.receive(JMSPagingFileDeleteTest.RECEIVE_TIMEOUT); - assertNotNull(message1); - Message message2 = subscriber2.receive(JMSPagingFileDeleteTest.RECEIVE_TIMEOUT); - assertNotNull(message2); + Assert.assertNotNull(subscriber1.receive(JMSPagingFileDeleteTest.RECEIVE_TIMEOUT)); + Assert.assertNotNull(subscriber2.receive(JMSPagingFileDeleteTest.RECEIVE_TIMEOUT)); } - pagingStore = server.getPagingManager().getPageStore(new SimpleString("jms.topic.topic1")); - long timeout = System.currentTimeMillis() + 5000; - while (timeout > System.currentTimeMillis() && pagingStore.isPaging()) - { - Thread.sleep(100); - } - assertFalse(pagingStore.isPaging()); - - printPageStoreInfo(pagingStore); + waitUntilPagingStops(5000); - assertEquals(0, pagingStore.getAddressSize()); + printPageStoreInfo(); + + Assert.assertEquals(0, getPagingStore().getAddressSize()); // assertEquals(1, pagingStore.getNumberOfPages()); //I expected number of the page is 1, but It was not. - assertFalse(pagingStore.isPaging()); // I expected IsPaging is false, but It was true. + Assert.assertFalse(getPagingStore().isPaging()); // I expected IsPaging is false, but It was true. // If the server is not restart, this test pass. // -----------------(Step4) Publish a message. the message is stored in the paging file. producer = session.createProducer(topic1); - bytesMessage = session.createBytesMessage(); - bytesMessage.writeBytes(new byte[JMSPagingFileDeleteTest.MESSAGE_SIZE]); - producer.send(bytesMessage); + sendMessage(createMessage(session), producer); - printPageStoreInfo(pagingStore); + printPageStoreInfo(); - assertEquals(1, pagingStore.getNumberOfPages()); //I expected number of the page is 1, but It was not. + Assert.assertEquals(1, getPagingStore().getNumberOfPages()); // I expected number of the page is 1, but It was + // not. } finally { + close(connection); + } + } + + public void testTopics_nonDurable() throws Exception + { + Connection connection = null; + + try + { + connection = cf.createConnection(); + + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + + MessageProducer producer = session.createProducer(topic1); + + printPageStoreInfo(); + + connection.start(); + + MessageConsumer subscriber = session.createConsumer(topic1); + final Message message = createMessage(session); + int numMessages = sendJustEnoughMessagesForPaging(message, producer); + + // ###Works if uncomment this to send one extra message or if use sendMessages instead above + // printPageStoreInfo(); + sendMessage(message, producer); + // numMessages++; + + printPageStoreInfo(); + + for (int i = 0; i < numMessages; i++) + { + Assert.assertNotNull(subscriber.receive(JMSPagingFileDeleteTest.RECEIVE_TIMEOUT)); + System.out.println("Messages recd:" + (i + 1)); + } + + assertNull(subscriber.receive(1000)); + + waitUntilPagingStops(5000); + + printPageStoreInfo(); + } + finally + { + close(connection); + } + } + + private void close(final Connection connection) + { + try + { if (connection != null) { connection.close(); } } + catch (JMSException e) + { + e.printStackTrace(); + } } - private void stopAndStartServer() throws Exception + private int sendMessages(final Message message, final MessageProducer producer) throws JMSException { - System.out.println("---------- Restart server. ----------"); - connection.close(); + System.out.println("---------- Send messages. ----------"); + for (int i = 0; i < JMSPagingFileDeleteTest.MESSAGE_NUM; i++) + { + sendMessage(message, producer); + } + System.out.println("Sent " + JMSPagingFileDeleteTest.MESSAGE_NUM + " messages."); - jmsServer.stop(); + return JMSPagingFileDeleteTest.MESSAGE_NUM; + } - jmsServer.start(); - jmsServer.activated(); - registerConnectionFactory(); + private int sendJustEnoughMessagesForPaging(final Message message, final MessageProducer producer) throws Exception + { + int messagesSendCount = 0; + while (!getPagingStore().isPaging()) + { + sendMessage(message, producer); + messagesSendCount++; + } - printPageStoreInfo(pagingStore); - reconnect(); + System.out.println(messagesSendCount + " messages sent before paging started"); + + return messagesSendCount; } - private void reconnect() throws Exception + private void sendMessage(final Message message, final MessageProducer producer) throws JMSException { - connection = cf.createConnection(); - connection.setClientID("cid"); - session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); - subscriber1 = session.createDurableSubscriber(topic1, "subscriber-1"); - subscriber2 = session.createDurableSubscriber(topic1, "subscriber-2"); - connection.start(); + producer.send(message); } - private void printPageStoreInfo(PagingStore pagingStore) throws Exception + private Message createMessage(final Session session) throws JMSException { + BytesMessage bytesMessage = session.createBytesMessage(); + bytesMessage.writeBytes(new byte[JMSPagingFileDeleteTest.MESSAGE_SIZE]); + return bytesMessage; + } + + private void waitUntilPagingStops(final int timeoutMillis) throws Exception, InterruptedException + { + long timeout = System.currentTimeMillis() + timeoutMillis; + while (timeout > System.currentTimeMillis() && getPagingStore().isPaging()) + { + Thread.sleep(100); + } + + if (!getPagingStore().isPaging()) + { + System.exit(-1); + } + Assert.assertFalse("Paging should have stopped", getPagingStore().isPaging()); + } + + private PagingStore getPagingStore() throws Exception + { + return server.getPagingManager().getPageStore(new SimpleString("jms.topic.topic1")); + } + + private void printPageStoreInfo() throws Exception + { + PagingStore pagingStore = getPagingStore(); System.out.println("---------- Paging Store Info ----------"); System.out.println(" CurrentPage = " + pagingStore.getCurrentPage()); System.out.println(" FirstPage = " + pagingStore.getFirstPage()); @@ -194,4 +269,17 @@ System.out.println(" Address Size = " + pagingStore.getAddressSize()); System.out.println(" Is Paging = " + pagingStore.isPaging()); } + + private void stopAndStartServer() throws Exception + { + System.out.println("---------- Restart server. ----------"); + + jmsServer.stop(); + + jmsServer.start(); + jmsServer.activated(); + registerConnectionFactory(); + + printPageStoreInfo(); + } } \ No newline at end of file Modified: branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/distribution/NettyFileStorageSymmetricClusterWithDiscoveryTest.java =================================================================== --- branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/distribution/NettyFileStorageSymmetricClusterWithDiscoveryTest.java 2011-08-01 16:25:50 UTC (rev 11090) +++ branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/distribution/NettyFileStorageSymmetricClusterWithDiscoveryTest.java 2011-08-01 19:39:43 UTC (rev 11091) @@ -13,6 +13,8 @@ package org.hornetq.tests.integration.cluster.distribution; +import org.hornetq.core.logging.Logger; + /** * A NettyFileStorageSymmetricClusterWithDiscoveryTest * @@ -22,6 +24,8 @@ */ public class NettyFileStorageSymmetricClusterWithDiscoveryTest extends SymmetricClusterWithDiscoveryTest { + Logger log = Logger.getLogger(NettyFileStorageSymmetricClusterWithDiscoveryTest.class); + @Override protected boolean isNetty() { @@ -32,6 +36,6 @@ { return true; } + - } Modified: branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/distribution/NettySymmetricClusterWithBackupTest.java =================================================================== --- branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/distribution/NettySymmetricClusterWithBackupTest.java 2011-08-01 16:25:50 UTC (rev 11090) +++ branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/distribution/NettySymmetricClusterWithBackupTest.java 2011-08-01 19:39:43 UTC (rev 11091) @@ -22,6 +22,8 @@ package org.hornetq.tests.integration.cluster.distribution; +import org.hornetq.core.logging.Logger; + /** * A NettySymmetricClusterWithBackupTest * @@ -31,6 +33,8 @@ */ public class NettySymmetricClusterWithBackupTest extends SymmetricClusterWithBackupTest { + private Logger log = Logger.getLogger(NettySymmetricClusterWithBackupTest.class); + @Override protected boolean isNetty() { @@ -39,10 +43,10 @@ public void _test() throws Exception { - for (int i = 0; i < 50; i++) + for (int i = 0; i < 500; i++) { - System.out.println("\n\n" + i + "\n\n"); - _testStartStopServers(); + log.info("#test " + i); + testNoLocalQueueLoadBalancedQueues (); tearDown(); setUp(); } Modified: branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/distribution/OneWayChainClusterTest.java =================================================================== --- branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/distribution/OneWayChainClusterTest.java 2011-08-01 16:25:50 UTC (rev 11090) +++ branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/distribution/OneWayChainClusterTest.java 2011-08-01 19:39:43 UTC (rev 11091) @@ -402,4 +402,18 @@ assertEquals(records.size(), 1); System.out.println("OneWayChainClusterTest.testChainClusterConnections"); } + + + public void _testLoop() throws Throwable + { + for (int i = 0 ; i < 500; i++) + { + log.info("#test " + i); + testRoundRobinForwardWhenNoConsumersTrue(); + tearDown(); + setUp(); + } + } + + } Modified: branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java =================================================================== --- branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java 2011-08-01 16:25:50 UTC (rev 11090) +++ branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java 2011-08-01 19:39:43 UTC (rev 11091) @@ -81,6 +81,17 @@ } }); } + + public void _testLoop() throws Throwable + { + for (int i = 0 ; i < 1000; i++) + { + log.info("#test " + i); + testTransactional(); + tearDown(); + setUp(); + } + } public void testTransactional() throws Throwable { @@ -366,10 +377,12 @@ } } - private void doTestTransactional(final TestRunner runner) throws Exception + private void doTestTransactional(final TestRunner runner) throws Throwable { // For duplication detection int executionId = 0; + + log.info("#test doTestTransactional starting now"); while (!runner.isFailed()) { @@ -411,10 +424,15 @@ addPayload(message); - + if (log.isDebugEnabled()) + { + log.debug("Sending message " + message); + } + producer.send(message); } + log.debug("Sending commit"); session.commit(); retry = false; @@ -423,32 +441,37 @@ { if (e.getCode() == HornetQException.DUPLICATE_ID_REJECTED) { + logAndSystemOut("#test duplicate id rejected"); break; } else if (e.getCode() == HornetQException.TRANSACTION_ROLLED_BACK || e.getCode() == HornetQException.UNBLOCKED) { + log.info("#test transaction rollback retrying"); // OK retry = true; } else { + log.info("#test Exception " + e, e); throw e; } } } while (retry); + logAndSystemOut("Finished sending, starting consumption now"); boolean blocked = false; retry = false; + ArrayList<Integer> msgs = new ArrayList<Integer>(); ClientConsumer consumer = null; do { - ArrayList<Integer> msgs = new ArrayList<Integer>(); + msgs.clear(); try { if (consumer == null) @@ -459,11 +482,16 @@ for (int i = 0; i < numMessages; i++) { - ClientMessage message = consumer.receive(500); + ClientMessage message = consumer.receive(10000); if (message == null) { break; } + + if (log.isDebugEnabled()) + { + log.debug("Received message " + message); + } int count = message.getIntProperty("counter"); @@ -473,14 +501,26 @@ } session.commit(); - - if (blocked) + + try { - assertTrue("msgs.size is expected to be 0 or " + numMessages + " but it was " + msgs.size(), msgs.size() == 0 || msgs.size() == numMessages); + if (blocked) + { + assertTrue("msgs.size is expected to be 0 or " + numMessages + " but it was " + msgs.size(), msgs.size() == 0 || msgs.size() == numMessages); + } + else + { + assertTrue("msgs.size is expected to be " + numMessages + " but it was " + msgs.size(), msgs.size() == numMessages); + } } - else + catch (Throwable e) { - assertTrue("msgs.size is expected to be " + numMessages + " but it was " + msgs.size(), msgs.size() == numMessages); + logAndSystemOut(e.getMessage() + " messages received"); + for (Integer msg : msgs) + { + logAndSystemOut(msg.toString()); + } + throw e; } int i = 0; @@ -496,6 +536,7 @@ { if (e.getCode() == HornetQException.TRANSACTION_ROLLED_BACK) { + logAndSystemOut("Transaction rolled back with " + msgs.size(), e); // TODO: https://jira.jboss.org/jira/browse/HORNETQ-369 // ATM RolledBack exception is being called with the transaction is committed. // the test will fail if you remove this next line @@ -503,6 +544,7 @@ } else if (e.getCode() == HornetQException.UNBLOCKED) { + logAndSystemOut("Unblocked with " + msgs.size(), e); // TODO: https://jira.jboss.org/jira/browse/HORNETQ-369 // This part of the test is never being called. blocked = true; @@ -514,6 +556,7 @@ } else { + logAndSystemOut(e.getMessage(), e); throw e; } } Modified: branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTestBase.java =================================================================== --- branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTestBase.java 2011-08-01 16:25:50 UTC (rev 11090) +++ branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTestBase.java 2011-08-01 19:39:43 UTC (rev 11091) @@ -192,6 +192,7 @@ @Override protected void tearDown() throws Exception { + logAndSystemOut("#test tearDown"); backupServer.stop(); liveServer.stop(); Modified: branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/failover/NettyAsynchronousReattachTest.java =================================================================== --- branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/failover/NettyAsynchronousReattachTest.java 2011-08-01 16:25:50 UTC (rev 11090) +++ branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/integration/cluster/failover/NettyAsynchronousReattachTest.java 2011-08-01 19:39:43 UTC (rev 11091) @@ -16,6 +16,7 @@ import org.hornetq.api.core.HornetQException; import org.hornetq.api.core.client.ClientSession; import org.hornetq.core.client.impl.ClientSessionInternal; +import org.hornetq.core.logging.Logger; /** * A NettyAsynchronousReattachTest @@ -40,12 +41,15 @@ // Package protected --------------------------------------------- // Protected ----------------------------------------------------- - + private final Logger log = Logger.getLogger(NettyAsynchronousReattachTest.class); + protected void crash(final ClientSession... sessions) throws Exception { + log.debug("Crashing sessions"); for (ClientSession session : sessions) { + log.debug("Crashing session " + session); ClientSessionInternal internalSession = (ClientSessionInternal) session; internalSession.getConnection().fail(new HornetQException(HornetQException.NOT_CONNECTED, "oops")); } Modified: branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/util/UnitTestCase.java =================================================================== --- branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/util/UnitTestCase.java 2011-08-01 16:25:50 UTC (rev 11090) +++ branches/Branch_2_2_EAP_cluster_clean2/tests/src/org/hornetq/tests/util/UnitTestCase.java 2011-08-01 19:39:43 UTC (rev 11091) @@ -363,7 +363,24 @@ return str.toString(); } + + /** Sends the message to both logger and System.out (for unit report) */ + public void logAndSystemOut(String message, Exception e) + { + Logger log = Logger.getLogger(this.getClass()); + log.info(message, e); + System.out.println(message); + e.printStackTrace(System.out); + } + /** Sends the message to both logger and System.out (for unit report) */ + public void logAndSystemOut(String message) + { + Logger log = Logger.getLogger(this.getClass()); + log.info(message); + System.out.println(message); + } + protected static TestSuite createAIOTestSuite(final Class<?> clazz) { TestSuite suite = new TestSuite(clazz.getName() + " testsuite");