You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by gt...@apache.org on 2015/03/02 15:56:56 UTC
[1/2] activemq git commit:
https://issues.apache.org/jira/browse/AMQ-5626 - fix and test. on restart,
next index needs to ingore priority to find the last used entry
Repository: activemq
Updated Branches:
refs/heads/master 4f5774493 -> ecebd2413
https://issues.apache.org/jira/browse/AMQ-5626 - fix and test. on restart, next index needs to ingore priority to find the last used entry
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/ecebd241
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/ecebd241
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/ecebd241
Branch: refs/heads/master
Commit: ecebd2413b878ba87710db9541f6bf353ceeb96e
Parents: 260e28e
Author: gtully <ga...@gmail.com>
Authored: Mon Mar 2 14:44:27 2015 +0000
Committer: gtully <ga...@gmail.com>
Committed: Mon Mar 2 14:44:58 2015 +0000
----------------------------------------------------------------------
.../activemq/store/kahadb/MessageDatabase.java | 34 ++-
.../activemq/store/kahadb/AMQ5626Test.java | 211 +++++++++++++++++++
2 files changed, 226 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/ecebd241/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
----------------------------------------------------------------------
diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
index 54188fb..188192f 100644
--- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
+++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
@@ -46,7 +46,6 @@ import java.util.Set;
import java.util.SortedSet;
import java.util.TreeMap;
import java.util.TreeSet;
-import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -2837,25 +2836,22 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
void configureLast(Transaction tx) throws IOException {
// Figure out the next key using the last entry in the destination.
- if (highPriorityIndex != null) {
- Entry<Long, MessageKeys> lastEntry = highPriorityIndex.getLast(tx);
- if (lastEntry != null) {
- nextMessageId = lastEntry.getKey() + 1;
- } else {
- lastEntry = defaultPriorityIndex.getLast(tx);
- if (lastEntry != null) {
- nextMessageId = lastEntry.getKey() + 1;
- } else {
- lastEntry = lowPriorityIndex.getLast(tx);
- if (lastEntry != null) {
- nextMessageId = lastEntry.getKey() + 1;
- }
- }
- }
- } else {
- Entry<Long, MessageKeys> lastEntry = defaultPriorityIndex.getLast(tx);
+ TreeSet<Long> orderedSet = new TreeSet<Long>();
+
+ addLast(orderedSet, highPriorityIndex, tx);
+ addLast(orderedSet, defaultPriorityIndex, tx);
+ addLast(orderedSet, lowPriorityIndex, tx);
+
+ if (!orderedSet.isEmpty()) {
+ nextMessageId = orderedSet.last() + 1;
+ }
+ }
+
+ private void addLast(TreeSet<Long> orderedSet, BTreeIndex<Long, MessageKeys> index, Transaction tx) throws IOException {
+ if (index != null) {
+ Entry<Long, MessageKeys> lastEntry = index.getLast(tx);
if (lastEntry != null) {
- nextMessageId = lastEntry.getKey() + 1;
+ orderedSet.add(lastEntry.getKey());
}
}
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/ecebd241/activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/AMQ5626Test.java
----------------------------------------------------------------------
diff --git a/activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/AMQ5626Test.java b/activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/AMQ5626Test.java
new file mode 100644
index 0000000..9097af7
--- /dev/null
+++ b/activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/AMQ5626Test.java
@@ -0,0 +1,211 @@
+/**
+ * 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.store.kahadb;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import javax.jms.ConnectionFactory;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.management.ObjectName;
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.TransportConnector;
+import org.apache.activemq.broker.jmx.DestinationView;
+import org.apache.activemq.broker.jmx.QueueView;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+
+
+public class AMQ5626Test {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AMQ5626Test.class);
+ private static final String QUEUE_NAME = "TesQ";
+ private BrokerService brokerService;
+ private URI brokerUri;
+
+ @Before
+ public void setup() throws Exception {
+ createBroker(true);
+ }
+
+ private void createBroker(boolean deleteMessagesOnStart) throws Exception {
+
+ brokerService = new BrokerService();
+
+ PolicyMap policyMap = new PolicyMap();
+ List<PolicyEntry> entries = new ArrayList<PolicyEntry>();
+ PolicyEntry pe = new PolicyEntry();
+ pe.setPrioritizedMessages(true);
+ pe.setExpireMessagesPeriod(0);
+
+ pe.setQueue(QUEUE_NAME);
+ entries.add(pe);
+
+ policyMap.setPolicyEntries(entries);
+
+ brokerService.setDestinationPolicy(policyMap);
+
+ TransportConnector transportConnector = new TransportConnector();
+ transportConnector.setName("openwire");
+ transportConnector.setUri(new URI("tcp://0.0.0.0:0"));
+ brokerService.addConnector(transportConnector);
+
+ brokerService.setDeleteAllMessagesOnStartup(deleteMessagesOnStart);
+ brokerService.getManagementContext().setCreateConnector(false);
+ brokerService.start();
+ brokerService.waitUntilStarted();
+
+ brokerUri = transportConnector.getPublishableConnectURI();
+ }
+
+ @Test
+ public void testPriorityMessages() throws Exception {
+
+ ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerUri);
+ ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();
+ connection.start();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+
+ MessageProducer producer = session.createProducer(session.createQueue(QUEUE_NAME));
+
+ Message message = session.createMessage();
+
+ // 0,1
+ producer.setPriority(9);
+ producer.send(message);
+ producer.send(message);
+
+ // 2,3
+ producer.setPriority(4);
+ producer.send(message);
+ producer.send(message);
+
+ connection.close();
+
+ stopRestartBroker();
+
+ connectionFactory = new ActiveMQConnectionFactory(brokerUri);
+ connection = (ActiveMQConnection) connectionFactory.createConnection();
+ connection.start();
+ session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ producer = session.createProducer(session.createQueue(QUEUE_NAME));
+
+ // 4
+ producer.setPriority(4);
+ producer.send(message);
+
+ displayQueueViews(brokerService);
+
+ // consume 5
+ MessageConsumer jmsConsumer = session.createConsumer(session.createQueue(QUEUE_NAME));
+ for (int i = 0; i < 5; i++) {
+ message = jmsConsumer.receive(4000);
+ assertNotNull("Got message i=" + i, message);
+ LOG.info("received: " + message.getJMSMessageID() + ", priority:" + message.getJMSPriority());
+ }
+
+ connection.close();
+ }
+
+ private void displayQueueViews(BrokerService broker) throws Exception {
+ Map<ObjectName, DestinationView> queueViews = broker.getAdminView().getBroker().getQueueViews();
+
+ for (ObjectName key : queueViews.keySet()) {
+ DestinationView destinationView = queueViews.get(key);
+
+ if (destinationView instanceof QueueView) {
+ QueueView queueView = (QueueView) destinationView;
+ LOG.info("ObjectName " + key);
+ LOG.info("QueueView name : " + queueView.getName());
+ LOG.info("QueueView cursorSize : " + queueView.cursorSize());
+ LOG.info("QueueView queueSize : " + queueView.getQueueSize());
+ LOG.info("QueueView enqueue count : " + queueView.getEnqueueCount());
+ LOG.info("QueueView dequeue count : " + queueView.getDequeueCount());
+ LOG.info("QueueView inflight count : " + queueView.getInFlightCount());
+
+ }
+ }
+ }
+
+ private QueueView getQueueView(BrokerService broker, String queueName) throws Exception {
+ Map<ObjectName, DestinationView> queueViews = broker.getAdminView().getBroker().getQueueViews();
+
+ for (ObjectName key : queueViews.keySet()) {
+ DestinationView destinationView = queueViews.get(key);
+
+ if (destinationView instanceof QueueView) {
+ QueueView queueView = (QueueView) destinationView;
+
+ if (queueView.getName().equals(queueName)) {
+ return queueView;
+ }
+
+ }
+ }
+ return null;
+ }
+
+ private synchronized void stopRestartBroker() {
+
+ try {
+
+ LOG.info(">>>SHUTTING BROKER DOWN");
+ brokerService.stop();
+ brokerService.waitUntilStopped();
+
+ //restart it
+ createBroker(false);
+ brokerService.start();
+ brokerService.waitUntilStarted();
+
+ LOG.info(">>>BROKER RESTARTED..");
+
+ } catch (Exception e) {
+ LOG.error("FAILED TO STOP/START BROKER EXCEPTION", e);
+ fail("FAILED TO STOP/START BROKER" + e);
+ }
+ }
+
+ @After
+ public void teardown() throws Exception {
+
+ try {
+ brokerService.stop();
+ } catch (Exception ex) {
+ LOG.error("FAILED TO STOP/START BROKER EXCEPTION", ex);
+ }
+
+ }
+
+}
\ No newline at end of file
[2/2] activemq git commit:
https://issues.apache.org/jira/browse/AMQ-5620 - avoid potential deadlock on
shutdown - waiting on connections to stop before stopping the pa would be an
alternative but may block for ever, auto rollback ensures there is no nee
Posted by gt...@apache.org.
https://issues.apache.org/jira/browse/AMQ-5620 - avoid potential deadlock on shutdown - waiting on connections to stop before stopping the pa would be an alternative but may block for ever, auto rollback ensures there is no need
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/260e28ec
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/260e28ec
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/260e28ec
Branch: refs/heads/master
Commit: 260e28ecadc654ce3f0a5b6c7058788235265574
Parents: 4f57744
Author: gtully <ga...@gmail.com>
Authored: Mon Mar 2 14:09:58 2015 +0000
Committer: gtully <ga...@gmail.com>
Committed: Mon Mar 2 14:44:58 2015 +0000
----------------------------------------------------------------------
.../store/kahadb/disk/journal/Journal.java | 29 ++++++++++++--------
1 file changed, 17 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/260e28ec/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/journal/Journal.java
----------------------------------------------------------------------
diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/journal/Journal.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/journal/Journal.java
index 91f82ae..54386cb 100644
--- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/journal/Journal.java
+++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/journal/Journal.java
@@ -463,20 +463,25 @@ public class Journal {
return dataFile.getNext();
}
- public synchronized void close() throws IOException {
- if (!started) {
- return;
- }
- if (this.timer != null) {
- this.timer.cancel();
+ public void close() throws IOException {
+ synchronized (this) {
+ if (!started) {
+ return;
+ }
+ if (this.timer != null) {
+ this.timer.cancel();
+ }
+ accessorPool.close();
}
- accessorPool.close();
+ // the appender can be calling back to to the journal blocking a close AMQ-5620
appender.close();
- fileMap.clear();
- fileByFileMap.clear();
- dataFiles.clear();
- lastAppendLocation.set(null);
- started = false;
+ synchronized (this) {
+ fileMap.clear();
+ fileByFileMap.clear();
+ dataFiles.clear();
+ lastAppendLocation.set(null);
+ started = false;
+ }
}
protected synchronized void cleanup() {