From a1a973ca340a408dad358ab250c24d0a73736f12 Mon Sep 17 00:00:00 2001 From: "Timothy A. Bish" Date: Thu, 13 Dec 2012 01:27:30 +0000 Subject: [PATCH] fix for: https://issues.apache.org/jira/browse/AMQ-4220 git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1421059 13f79535-47bb-0310-9956-ffa450edef68 --- .../org/apache/activemq/bugs/AMQ4220Test.java | 120 ++++++++++++++++++ .../kahadb/MultiKahaDBPersistenceAdapter.java | 5 +- 2 files changed, 122 insertions(+), 3 deletions(-) create mode 100644 activemq-core/src/test/java/org/apache/activemq/bugs/AMQ4220Test.java diff --git a/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ4220Test.java b/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ4220Test.java new file mode 100644 index 0000000000..7084bde44c --- /dev/null +++ b/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ4220Test.java @@ -0,0 +1,120 @@ +/** + * 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.activemq.bugs; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.util.ArrayList; + +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.command.ActiveMQQueue; +import org.apache.activemq.store.PersistenceAdapter; +import org.apache.activemq.store.kahadb.FilteredKahaDBPersistenceAdapter; +import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter; +import org.apache.activemq.store.kahadb.MultiKahaDBPersistenceAdapter; +import org.apache.activemq.util.Wait; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class AMQ4220Test { + + static final Logger LOG = LoggerFactory.getLogger(AMQ4220Test.class); + private final static int maxFileLength = 1024*1024*32; + private final static String destinationName = "TEST.QUEUE"; + BrokerService broker; + + @Before + public void setUp() throws Exception { + prepareBrokerWithMultiStore(true); + broker.start(); + broker.waitUntilStarted(); + } + + @After + public void tearDown() throws Exception { + broker.stop(); + } + + protected BrokerService createBroker(PersistenceAdapter kaha) throws Exception { + BrokerService broker = new BrokerService(); + broker.setUseJmx(true); + broker.setBrokerName("localhost"); + broker.setPersistenceAdapter(kaha); + return broker; + } + + @Test + public void testRestartAfterQueueDelete() throws Exception { + + // Ensure we have an Admin View. + assertTrue("Broker doesn't have an Admin View.", Wait.waitFor(new Wait.Condition() { + @Override + public boolean isSatisified() throws Exception { + return (broker.getAdminView()) != null; + } + })); + + + LOG.info("Adding initial destination: {}", destinationName); + + broker.getAdminView().addQueue(destinationName); + + assertNotNull(broker.getDestination(new ActiveMQQueue(destinationName))); + + LOG.info("Removing initial destination: {}", destinationName); + + broker.getAdminView().removeQueue(destinationName); + + LOG.info("Adding back destination: {}", destinationName); + + broker.getAdminView().addQueue(destinationName); + + assertNotNull(broker.getDestination(new ActiveMQQueue(destinationName))); + } + + protected KahaDBPersistenceAdapter createStore(boolean delete) throws IOException { + KahaDBPersistenceAdapter kaha = new KahaDBPersistenceAdapter(); + kaha.setJournalMaxFileLength(maxFileLength); + kaha.setCleanupInterval(5000); + if (delete) { + kaha.deleteAllMessages(); + } + return kaha; + } + + public void prepareBrokerWithMultiStore(boolean deleteAllMessages) throws Exception { + + MultiKahaDBPersistenceAdapter multiKahaDBPersistenceAdapter = new MultiKahaDBPersistenceAdapter(); + if (deleteAllMessages) { + multiKahaDBPersistenceAdapter.deleteAllMessages(); + } + ArrayList adapters = new ArrayList(); + + FilteredKahaDBPersistenceAdapter template = new FilteredKahaDBPersistenceAdapter(); + template.setPersistenceAdapter(createStore(deleteAllMessages)); + template.setPerDestination(true); + adapters.add(template); + + multiKahaDBPersistenceAdapter.setFilteredPersistenceAdapters(adapters); + broker = createBroker(multiKahaDBPersistenceAdapter); + } +} diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MultiKahaDBPersistenceAdapter.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MultiKahaDBPersistenceAdapter.java index 0ac4e03a03..f3870f4fab 100644 --- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MultiKahaDBPersistenceAdapter.java +++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MultiKahaDBPersistenceAdapter.java @@ -124,7 +124,6 @@ public class MultiKahaDBPersistenceAdapter extends DestinationMap implements Per } private String nameFromDestinationFilter(ActiveMQDestination destination) { - if (destination.getQualifiedName().length() > IOHelper.getMaxFileNameLength()) { LOG.warn("Destination name is longer than 'MaximumFileNameLength' system property, " + "potential problem with recovery can result from name truncation."); @@ -247,10 +246,10 @@ public class MultiKahaDBPersistenceAdapter extends DestinationMap implements Per @Override public void removeQueueMessageStore(ActiveMQQueue destination) { PersistenceAdapter adapter = getMatchingPersistenceAdapter(destination); - adapter.removeQueueMessageStore(destination); if (adapter instanceof KahaDBPersistenceAdapter) { adapter.removeQueueMessageStore(destination); removeMessageStore((KahaDBPersistenceAdapter)adapter, destination); + removeAll(destination); } } @@ -260,6 +259,7 @@ public class MultiKahaDBPersistenceAdapter extends DestinationMap implements Per if (adapter instanceof KahaDBPersistenceAdapter) { adapter.removeTopicMessageStore(destination); removeMessageStore((KahaDBPersistenceAdapter)adapter, destination); + removeAll(destination); } } @@ -464,5 +464,4 @@ public class MultiKahaDBPersistenceAdapter extends DestinationMap implements Per String path = getDirectory() != null ? getDirectory().getAbsolutePath() : "DIRECTORY_NOT_SET"; return "MultiKahaDBPersistenceAdapter[" + path + "]" + adapters; } - }