Author: jbertram Date: 2012-02-10 13:51:28 -0500 (Fri, 10 Feb 2012) New Revision: 12108 Added: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/client/CreateQueueIdempotentTest.java Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/StorageManager.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/postoffice/impl/SimpleAddressManager.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java Log: [HORNETQ-847] createQueue not idempotent Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/StorageManager.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/StorageManager.java 2012-02-10 17:03:56 UTC (rev 12107) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/StorageManager.java 2012-02-10 18:51:28 UTC (rev 12108) @@ -193,6 +193,10 @@ void commit(long txID, boolean lineUpContext) throws Exception; void rollback(long txID) throws Exception; + + void rollbackBindings(long txID) throws Exception; + + void commitBindings(long txID) throws Exception; void storePageTransaction(long txID, PageTransactionInfo pageTransaction) throws Exception; @@ -220,7 +224,7 @@ // Bindings related operations - void addQueueBinding(Binding binding) throws Exception; + void addQueueBinding(long tx, Binding binding) throws Exception; void deleteQueueBinding(long queueBindingID) throws Exception; Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2012-02-10 17:03:56 UTC (rev 12107) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2012-02-10 18:51:28 UTC (rev 12108) @@ -758,6 +758,17 @@ commit(txID, true); } + public void commitBindings(final long txID) throws Exception + { + bindingsJournal.appendCommitRecord(txID, true); + } + + public void rollbackBindings(final long txID) throws Exception + { + // no need to sync, it's going away anyways + bindingsJournal.appendRollbackRecord(txID, false); + } + public void commit(final long txID, final boolean lineUpContext) throws Exception { messageJournal.appendCommitRecord(txID, syncTransactional, getContext(syncTransactional), lineUpContext); @@ -1371,7 +1382,7 @@ // Bindings operations - public void addQueueBinding(final Binding binding) throws Exception + public void addQueueBinding(final long tx, final Binding binding) throws Exception { Queue queue = (Queue)binding.getBindable(); @@ -1383,10 +1394,9 @@ binding.getAddress(), filterString); - bindingsJournal.appendAddRecord(binding.getID(), + bindingsJournal.appendAddRecordTransactional(tx, binding.getID(), JournalStorageManager.QUEUE_BINDING_RECORD, - bindingEncoding, - true); + bindingEncoding); } public void deleteQueueBinding(final long queueBindingID) throws Exception Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2012-02-10 17:03:56 UTC (rev 12107) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2012-02-10 18:51:28 UTC (rev 12108) @@ -13,16 +13,6 @@ package org.hornetq.core.persistence.impl.nullpm; -import java.nio.ByteBuffer; -import java.util.Collections; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.Executor; -import java.util.concurrent.atomic.AtomicLong; - -import javax.transaction.xa.Xid; - import org.hornetq.api.core.Pair; import org.hornetq.api.core.SimpleString; import org.hornetq.core.journal.IOAsyncTask; @@ -49,6 +39,15 @@ import org.hornetq.core.transaction.ResourceManager; import org.hornetq.core.transaction.Transaction; +import javax.transaction.xa.Xid; +import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.Executor; +import java.util.concurrent.atomic.AtomicLong; + /** * * A NullStorageManager @@ -140,10 +139,18 @@ { } - public void storeReference(final long queueID, final long messageID, final boolean last) throws Exception + public void rollbackBindings(long txID) throws Exception { } + public void commitBindings(long txID) throws Exception + { + } + + public void storeReference(final long queueID, final long messageID, final boolean last) throws Exception + { + } + public void storeReferenceTransactional(final long txID, final long queueID, final long messageID) throws Exception { } @@ -227,7 +234,11 @@ { } - /* (non-Javadoc) + public void addQueueBinding(long tx, Binding binding) throws Exception + { + } + + /* (non-Javadoc) * @see org.hornetq.core.persistence.StorageManager#createLargeMessageStorage(long, int, int) */ public LargeServerMessage createLargeMessage() Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/postoffice/impl/SimpleAddressManager.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/postoffice/impl/SimpleAddressManager.java 2012-02-10 17:03:56 UTC (rev 12107) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/postoffice/impl/SimpleAddressManager.java 2012-02-10 18:51:28 UTC (rev 12108) @@ -16,6 +16,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import org.hornetq.api.core.HornetQException; import org.hornetq.api.core.SimpleString; import org.hornetq.core.logging.Logger; import org.hornetq.core.postoffice.Address; @@ -56,7 +57,7 @@ { if (nameMap.putIfAbsent(binding.getUniqueName(), binding) != null) { - throw new IllegalStateException("Binding already exists " + binding); + throw new HornetQException(HornetQException.QUEUE_EXISTS, "Binding already exists " + binding); } if (log.isDebugEnabled()) Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2012-02-10 17:03:56 UTC (rev 12107) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2012-02-10 18:51:28 UTC (rev 12108) @@ -1682,6 +1682,7 @@ Filter filter = FilterImpl.createFilter(filterString); + long txID = storageManager.generateUniqueID();; long queueID = storageManager.generateUniqueID(); PageSubscription pageSubscription; @@ -1709,10 +1710,28 @@ if (durable) { - storageManager.addQueueBinding(binding); + storageManager.addQueueBinding(txID, binding); } - postOffice.addBinding(binding); + try + { + postOffice.addBinding(binding); + if (durable) + { + storageManager.commitBindings(txID); + } + } + catch (Exception e) + { + if (durable) + { + storageManager.rollbackBindings(txID); + } + queue.close(); + pageSubscription.close(); + throw e; + } + managementService.registerAddress(address); managementService.registerQueue(queue, address, storageManager); Added: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/client/CreateQueueIdempotentTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/client/CreateQueueIdempotentTest.java (rev 0) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/client/CreateQueueIdempotentTest.java 2012-02-10 18:51:28 UTC (rev 12108) @@ -0,0 +1,211 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat 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.hornetq.tests.integration.client; + +import junit.framework.Assert; +import org.hornetq.api.core.HornetQException; +import org.hornetq.api.core.SimpleString; +import org.hornetq.api.core.TransportConfiguration; +import org.hornetq.api.core.client.ClientSession; +import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.api.core.client.HornetQClient; +import org.hornetq.api.core.client.ServerLocator; +import org.hornetq.core.config.Configuration; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.server.HornetQServer; +import org.hornetq.core.server.HornetQServers; +import org.hornetq.tests.util.ServiceTestBase; + +import java.util.concurrent.atomic.AtomicInteger; + +public class CreateQueueIdempotentTest extends ServiceTestBase +{ + private static final Logger log = Logger.getLogger(CreateQueueIdempotentTest.class); + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + public void testSequentialCreateQueueIdempotency() throws Exception + { + boolean success = false; + final SimpleString QUEUE = new SimpleString("SequentialCreateQueueIdempotency"); + + Configuration conf = createDefaultConfig(); + + conf.setSecurityEnabled(false); + + conf.getAcceptorConfigurations().add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory")); + + HornetQServer server = HornetQServers.newHornetQServer(conf, true); + + server.start(); + ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); + + ClientSessionFactory sf = locator.createSessionFactory(); + + ClientSession session = sf.createSession(false, true, true); + + session.createQueue(QUEUE, QUEUE, null, true); + + try + { + session.createQueue(QUEUE, QUEUE, null, true); + } + catch (Exception e) + { + if (e instanceof HornetQException) + { + if (((HornetQException) e).getCode() == 101) + { + success = true; + } + } + } + + session.close(); + + locator.close(); + + server.stop(); + + Assert.assertTrue(success); + } + + public void testConcurrentCreateQueueIdempotency() throws Exception + { + boolean success = true; + final String QUEUE = "ConcurrentCreateQueueIdempotency"; + AtomicInteger queuesCreated = new AtomicInteger(0); + AtomicInteger failedAttempts = new AtomicInteger(0); + + Configuration conf = createDefaultConfig(); + + conf.setSecurityEnabled(false); + + conf.getAcceptorConfigurations().add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory")); + + HornetQServer server = HornetQServers.newHornetQServer(conf, true); + + server.start(); + final int NUM_THREADS = 5; + + QueueCreator[] queueCreators = new QueueCreator[NUM_THREADS]; + + + for(int i = 0; i < NUM_THREADS; i++) + { + QueueCreator queueCreator = new QueueCreator(QUEUE, queuesCreated, failedAttempts); + queueCreators[i] = queueCreator; + } + + for(int i = 0; i < NUM_THREADS; i++) + { + queueCreators[i].start(); + } + + for(int i = 0; i < NUM_THREADS; i++) + { + queueCreators[i].join(); + } + + server.stop(); + + // re-starting the server appears to be an unreliable guide + try + { + server.start(); + } catch (Exception e) + { + System.out.println("THIS BLEW UP!!"); + e.printStackTrace(); + success = false; + } + + server.stop(); + + Assert.assertTrue(success); + Assert.assertEquals(1, queuesCreated.intValue()); + Assert.assertEquals(NUM_THREADS - 1, failedAttempts.intValue()); + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + + class QueueCreator extends Thread + { + private String queueName = null; + private AtomicInteger queuesCreated = null; + private AtomicInteger failedAttempts = null; + + + QueueCreator(String queueName, AtomicInteger queuesCreated, AtomicInteger failedAttempts) + { + this.queueName = queueName; + this.queuesCreated = queuesCreated; + this.failedAttempts = failedAttempts; + } + @Override + public void run() + { + ServerLocator locator = null; + ClientSession session = null; + + try + { + locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); + ClientSessionFactory sf = locator.createSessionFactory(); + session = sf.createSession(false, true, true); + final SimpleString QUEUE = new SimpleString(queueName); + session.createQueue(QUEUE, QUEUE, null, true); + queuesCreated.incrementAndGet(); + } + catch (Exception e) + { + e.printStackTrace(); + if (e instanceof HornetQException) + { + if (((HornetQException) e).getCode() == 101) + { + failedAttempts.incrementAndGet(); + } + } + } + finally + { + if (locator != null) { + locator.close(); + } + if (session != null) { + try { + session.close(); + } catch (HornetQException e) { + e.printStackTrace(); + } + } + } + } + } +} Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2012-02-10 17:03:56 UTC (rev 12107) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2012-02-10 18:51:28 UTC (rev 12108) @@ -13,26 +13,7 @@ package org.hornetq.tests.unit.core.paging.impl; -import java.nio.ByteBuffer; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.Executor; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicLong; - -import javax.transaction.xa.Xid; - import junit.framework.Assert; - import org.hornetq.api.core.HornetQBuffer; import org.hornetq.api.core.HornetQBuffers; import org.hornetq.api.core.Pair; @@ -43,12 +24,7 @@ import org.hornetq.core.journal.SequentialFileFactory; import org.hornetq.core.journal.impl.NIOSequentialFileFactory; import org.hornetq.core.message.impl.MessageInternal; -import org.hornetq.core.paging.Page; -import org.hornetq.core.paging.PageTransactionInfo; -import org.hornetq.core.paging.PagedMessage; -import org.hornetq.core.paging.PagingManager; -import org.hornetq.core.paging.PagingStore; -import org.hornetq.core.paging.PagingStoreFactory; +import org.hornetq.core.paging.*; import org.hornetq.core.paging.cursor.PagePosition; import org.hornetq.core.paging.impl.PageTransactionInfoImpl; import org.hornetq.core.paging.impl.PagingStoreImpl; @@ -81,6 +57,13 @@ import org.hornetq.tests.util.UnitTestCase; import org.hornetq.utils.ExecutorFactory; +import javax.transaction.xa.Xid; +import java.nio.ByteBuffer; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + /** * * @author <a href="mailto:clebert.suconic@jboss.com">Clebert Suconic</a> @@ -1254,7 +1237,15 @@ { } - /* (non-Javadoc) + public void rollbackBindings(long txID) throws Exception + { + } + + public void commitBindings(long txID) throws Exception + { + } + + /* (non-Javadoc) * @see org.hornetq.core.persistence.StorageManager#storeAcknowledge(long, long) */ public void storeAcknowledge(final long queueID, final long messageID) throws Exception @@ -1329,7 +1320,11 @@ { } - /* (non-Javadoc) + public void addQueueBinding(long tx, Binding binding) throws Exception + { + } + + /* (non-Javadoc) * @see org.hornetq.core.persistence.StorageManager#updateDeliveryCount(org.hornetq.core.server.MessageReference) */ public void updateDeliveryCount(final MessageReference ref) throws Exception