You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ha...@apache.org on 2014/03/13 16:49:01 UTC
[01/13] git commit: Added test to reproduce AMQ-4887
Repository: activemq
Updated Branches:
refs/heads/activemq-5.9 687aa9265 -> 88959d5c6
Added test to reproduce AMQ-4887
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/bbf5c674
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/bbf5c674
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/bbf5c674
Branch: refs/heads/activemq-5.9
Commit: bbf5c67489057dacfc7127dd69a0cf5584291f13
Parents: 687aa92
Author: Kevin Earls <ke...@kevinearls.com>
Authored: Thu Dec 5 10:45:38 2013 +0100
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Thu Mar 13 11:04:43 2014 -0400
----------------------------------------------------------------------
.../org/apache/activemq/bugs/AMQ4887Test.java | 81 ++++++++++++++++++++
1 file changed, 81 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/bbf5c674/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4887Test.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4887Test.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4887Test.java
new file mode 100644
index 0000000..f29f605
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4887Test.java
@@ -0,0 +1,81 @@
+/**
+ * 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 org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.BytesMessage;
+import javax.jms.Destination;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+
+import static org.junit.Assert.*;
+
+public class AMQ4887Test {
+ private static final transient Logger LOG = LoggerFactory.getLogger(AMQ4887Test.class);
+ private static final Integer ITERATIONS = 10;
+ @Rule
+ public TestName name = new TestName();
+
+ @Test
+ public void testSetPropertyBeforeCopy() throws Exception {
+ ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");
+ ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();
+ connection.start();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Destination destination = session.createQueue(name.toString());
+ MessageConsumer consumer = session.createConsumer(destination);
+ MessageProducer producer = session.createProducer(destination);
+
+ byte[] messageContent = "bytes message".getBytes();
+ BytesMessage bytesMessage = session.createBytesMessage();
+ bytesMessage.writeBytes(messageContent);
+
+ for (int i=0; i < ITERATIONS; i++) {
+ long sendTime = System.currentTimeMillis();
+ bytesMessage.setLongProperty("sendTime", sendTime);
+ producer.send(bytesMessage);
+
+ LOG.debug("Receiving message " + i);
+ Message receivedMessage = consumer.receive(5000);
+ assertNotNull("On message " + i, receivedMessage);
+ assertTrue("On message " + i, receivedMessage instanceof BytesMessage);
+
+ BytesMessage receivedBytesMessage = (BytesMessage) receivedMessage;
+ byte[] receivedBytes = new byte[(int) receivedBytesMessage.getBodyLength()];
+ receivedBytesMessage.readBytes(receivedBytes);
+ LOG.debug("Message " + i + " content [" + new String(receivedBytes) + "]");
+ assertEquals("On message " + i, messageContent.length, receivedBytes.length);
+ assertArrayEquals("On message " + i, messageContent, receivedBytes);
+
+ long receivedSendTime = receivedBytesMessage.getLongProperty("sendTime");
+ assertEquals("On message " + i, receivedSendTime, sendTime);
+
+ Thread.sleep(10);
+ }
+ }
+
+}
[08/13] git commit: added time unit for "send timeout" in the javadocs
Posted by ha...@apache.org.
added time unit for "send timeout" in the javadocs
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/4a80e958
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/4a80e958
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/4a80e958
Branch: refs/heads/activemq-5.9
Commit: 4a80e958447eadfc82ad24be26903aee1b833a05
Parents: f713c82
Author: Christian Posta <ch...@gmail.com>
Authored: Mon Dec 9 09:14:38 2013 -0700
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Thu Mar 13 11:06:42 2014 -0400
----------------------------------------------------------------------
.../src/main/java/org/apache/activemq/ActiveMQConnection.java | 4 ++--
.../main/java/org/apache/activemq/ActiveMQConnectionFactory.java | 4 ++--
2 files changed, 4 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/4a80e958/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java
----------------------------------------------------------------------
diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java
index 5057323..ec4d38a 100755
--- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java
+++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java
@@ -1777,14 +1777,14 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
}
/**
- * @return the sendTimeout
+ * @return the sendTimeout (in milliseconds)
*/
public int getSendTimeout() {
return sendTimeout;
}
/**
- * @param sendTimeout the sendTimeout to set
+ * @param sendTimeout the sendTimeout to set (in milliseconds)
*/
public void setSendTimeout(int sendTimeout) {
this.sendTimeout = sendTimeout;
http://git-wip-us.apache.org/repos/asf/activemq/blob/4a80e958/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
----------------------------------------------------------------------
diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
index fbbfc2b..58fee6e 100755
--- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
+++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
@@ -616,14 +616,14 @@ public class ActiveMQConnectionFactory extends JNDIBaseStorable implements Conne
}
/**
- * @return the sendTimeout
+ * @return the sendTimeout (in milliseconds)
*/
public int getSendTimeout() {
return sendTimeout;
}
/**
- * @param sendTimeout the sendTimeout to set
+ * @param sendTimeout the sendTimeout to set (in milliseconds)
*/
public void setSendTimeout(int sendTimeout) {
this.sendTimeout = sendTimeout;
[02/13] git commit: Test to reproduce AMQ4920
Posted by ha...@apache.org.
Test to reproduce AMQ4920
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/af09bdc5
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/af09bdc5
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/af09bdc5
Branch: refs/heads/activemq-5.9
Commit: af09bdc5964f1778e891064563e7c01b454d4b8f
Parents: bbf5c67
Author: Kevin Earls <ke...@kevinearls.com>
Authored: Thu Dec 5 12:50:55 2013 +0100
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Thu Mar 13 11:04:47 2014 -0400
----------------------------------------------------------------------
.../activemq/transport/amqp/AMQ4920Test.java | 162 +++++++++++++++++++
1 file changed, 162 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/af09bdc5/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AMQ4920Test.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AMQ4920Test.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AMQ4920Test.java
new file mode 100644
index 0000000..5d6d473
--- /dev/null
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AMQ4920Test.java
@@ -0,0 +1,162 @@
+/**
+ * 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.transport.amqp;
+
+import org.apache.qpid.amqp_1_0.jms.impl.ConnectionFactoryImpl;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+
+public class AMQ4920Test extends AmqpTestSupport {
+ private static final Logger LOG = LoggerFactory.getLogger(AMQ4920Test.class);
+ private static final Integer ITERATIONS = 1 * 1000;
+ private static final Integer CONSUMER_COUNT = 4; // At least 2 consumers are required to reproduce the original issue
+ public static final String TEXT_MESSAGE = "TextMessage: ";
+ private CountDownLatch latch = new CountDownLatch(CONSUMER_COUNT * ITERATIONS);
+
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ this.autoFailTestSupport.setAutoFail(false);
+ }
+
+ @Test(timeout = 5 * 60 * 1000)
+ public void testSendWithMultipleConsumers() throws Exception {
+ ConnectionFactory connectionFactory = (ConnectionFactory) new ConnectionFactoryImpl("localhost", port, "admin", "admin");
+ Connection connection = connectionFactory.createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ String destinationName = "topic://AMQ4920Test" + System.currentTimeMillis();
+ Destination destination = session.createTopic(destinationName);
+ connection.start();
+
+ ExecutorService executor = Executors.newCachedThreadPool();
+ for (int i=0; i < CONSUMER_COUNT; i++) {
+ AMQ4930ConsumerTask consumerTask = new AMQ4930ConsumerTask(destinationName, port, "Consumer-" + i, latch, ITERATIONS);
+ executor.submit(consumerTask);
+ }
+ connection.start();
+
+ LOG.debug("At start latch is " + latch.getCount());
+ sendMessages(connection, destination, ITERATIONS, 10);
+ LOG.debug("After send latch is " + latch.getCount());
+
+ latch.await(15, TimeUnit.SECONDS);
+ LOG.debug("After await latch is " + latch.getCount());
+ assertEquals(0, latch.getCount());
+
+ executor.shutdown();
+ }
+
+ public void sendMessages(Connection connection, Destination destination, int count, int sleepInterval) throws Exception {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(destination);
+
+ for (int i = 0; i < count; i++) {
+ TextMessage message = session.createTextMessage();
+ message.setText(TEXT_MESSAGE + i);
+ LOG.debug("Sending message [" + i + "]");
+ producer.send(message);
+ if (sleepInterval > 0) {
+ Thread.sleep(sleepInterval);
+ }
+ }
+
+ session.close();
+ }
+}
+
+
+class AMQ4930ConsumerTask implements Callable<Boolean> {
+ protected static final Logger LOG = LoggerFactory.getLogger(AMQ4930ConsumerTask.class);
+ private String destinationName;
+ private String consumerName;
+ private CountDownLatch messagesReceived;
+ private int port;
+ private int expectedMessageCount;
+
+ public AMQ4930ConsumerTask (String destinationName, int port, String consumerName, CountDownLatch latch, int expectedMessageCount) {
+ this.destinationName = destinationName;
+ this.port = port;
+ this.consumerName = consumerName;
+ this.messagesReceived = latch;
+ this.expectedMessageCount = expectedMessageCount;
+ }
+
+ @Override
+ public Boolean call() throws Exception {
+ LOG.debug(consumerName + " starting");
+ Connection connection=null;
+ try {
+ ConnectionFactory connectionFactory = (ConnectionFactory) new ConnectionFactoryImpl("localhost", port, "admin", "admin");
+ connection = connectionFactory.createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Destination destination = session.createTopic(destinationName);
+ MessageConsumer consumer = session.createConsumer(destination);
+ connection.start();
+
+ int receivedCount = 0;
+ while(receivedCount < expectedMessageCount) {
+ Message message = consumer.receive(5 * 1000);
+ if (message == null) {
+ LOG.error("consumer {} got null message on iteration {}", consumerName, receivedCount);
+ return false;
+ }
+ if (!(message instanceof TextMessage)) {
+ LOG.error("consumer {} expected text message on iteration {} but got {}", consumerName, receivedCount, message.getClass().getCanonicalName());
+ return false;
+ }
+ TextMessage tm = (TextMessage) message;
+ if (!tm.getText().equals(AMQ4920Test.TEXT_MESSAGE + receivedCount)) {
+ LOG.error("consumer {} expected {} got message [{}]", consumerName, receivedCount, tm.getText());
+ return false;
+ }
+ LOG.debug("consumer {} expected {} got message [{}]", consumerName, receivedCount, tm.getText()); // TODO make debug
+
+ messagesReceived.countDown();
+ receivedCount++;
+ }
+ } catch (Exception e) {
+ LOG.error("UnexpectedException in " + consumerName, e);
+ } finally {
+ try {
+ connection.close();
+ } catch (JMSException ignoreMe) {
+ }
+ }
+
+ return true;
+ }
+}
+
[12/13] git commit: https://issues.apache.org/jira/browse/AMQ-4926 -
karaf version range
Posted by ha...@apache.org.
https://issues.apache.org/jira/browse/AMQ-4926 - karaf version range
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/6d236b09
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/6d236b09
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/6d236b09
Branch: refs/heads/activemq-5.9
Commit: 6d236b097644d01a6402d73b17d9e08628aa3e18
Parents: cdbd622
Author: Dejan Bosanac <de...@nighttale.net>
Authored: Tue Dec 10 11:20:58 2013 +0100
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Thu Mar 13 11:08:00 2014 -0400
----------------------------------------------------------------------
activemq-karaf/pom.xml | 2 ++
1 file changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/6d236b09/activemq-karaf/pom.xml
----------------------------------------------------------------------
diff --git a/activemq-karaf/pom.xml b/activemq-karaf/pom.xml
index 3ed4840..229684e 100644
--- a/activemq-karaf/pom.xml
+++ b/activemq-karaf/pom.xml
@@ -171,6 +171,8 @@
<Export-Package>org.apache.activemq.karaf*;version=${project.version};-split-package:=merge-first</Export-Package>
<Import-Package>
org.apache.felix.gogo.commands,
+ org.apache.karaf.shell.console;version="[2,4)",
+ org.apache.karaf.shell.console.commands;version="[2,4)",
*
</Import-Package>
<Private-Package>!*</Private-Package>
[06/13] git commit: Default flushDelay on the replicated leveldb
store did not have the correct default.
Posted by ha...@apache.org.
Default flushDelay on the replicated leveldb store did not have the correct default.
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/7f68893a
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/7f68893a
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/7f68893a
Branch: refs/heads/activemq-5.9
Commit: 7f68893af6bbbec097a5a47a3196c1b352771106
Parents: bc2f708
Author: Hiram Chirino <hi...@hiramchirino.com>
Authored: Thu Dec 5 13:33:29 2013 -0500
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Thu Mar 13 11:06:13 2014 -0400
----------------------------------------------------------------------
.../apache/activemq/leveldb/replicated/ElectingLevelDBStore.scala | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/7f68893a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/ElectingLevelDBStore.scala
----------------------------------------------------------------------
diff --git a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/ElectingLevelDBStore.scala b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/ElectingLevelDBStore.scala
index 55e0419..4ff4288 100644
--- a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/ElectingLevelDBStore.scala
+++ b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/ElectingLevelDBStore.scala
@@ -124,7 +124,7 @@ class ElectingLevelDBStore extends ProxyLevelDBStore {
@BeanProperty
var indexCacheSize: Long = 1024 * 1024 * 256L
@BeanProperty
- var flushDelay = 1000 * 5
+ var flushDelay = 0
@BeanProperty
var asyncBufferSize = 1024 * 1024 * 4
@BeanProperty
[04/13] git commit: fix for:
https://issues.apache.org/jira/browse/AMQ-4910
Posted by ha...@apache.org.
fix for: https://issues.apache.org/jira/browse/AMQ-4910
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/d26545c7
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/d26545c7
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/d26545c7
Branch: refs/heads/activemq-5.9
Commit: d26545c73581b9898a3c4b52e84af5c457be07e1
Parents: 6a7e640
Author: Timothy Bish <ta...@gmai.com>
Authored: Thu Dec 5 12:01:44 2013 -0500
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Thu Mar 13 11:05:01 2014 -0400
----------------------------------------------------------------------
assembly/src/release/bin/win32/wrapper.conf | 2 +-
assembly/src/release/bin/win64/wrapper.conf | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/d26545c7/assembly/src/release/bin/win32/wrapper.conf
----------------------------------------------------------------------
diff --git a/assembly/src/release/bin/win32/wrapper.conf b/assembly/src/release/bin/win32/wrapper.conf
index d771387..45ade30 100644
--- a/assembly/src/release/bin/win32/wrapper.conf
+++ b/assembly/src/release/bin/win32/wrapper.conf
@@ -59,7 +59,7 @@ wrapper.java.additional.11=-Dactivemq.data="%ACTIVEMQ_DATA%"
wrapper.java.additional.12=-Dhawtio.realm=activemq
wrapper.java.additional.13=-Dhawtio.role=admins
wrapper.java.additional.14=-Dhawtio.rolePrincipalClasses=org.apache.activemq.jaas.GroupPrincipal
-wrapper.java.additional.15=-Djava.security.auth.login.config=%ACTIVEMQ_CONF%/login.config
+wrapper.java.additional.15=-Djava.security.auth.login.config="%ACTIVEMQ_CONF%/login.config"
# Uncomment to enable remote jmx
#wrapper.java.additional.n=-Dcom.sun.management.jmxremote.port=1616
http://git-wip-us.apache.org/repos/asf/activemq/blob/d26545c7/assembly/src/release/bin/win64/wrapper.conf
----------------------------------------------------------------------
diff --git a/assembly/src/release/bin/win64/wrapper.conf b/assembly/src/release/bin/win64/wrapper.conf
index 2ca5b2b..35c00b5 100644
--- a/assembly/src/release/bin/win64/wrapper.conf
+++ b/assembly/src/release/bin/win64/wrapper.conf
@@ -59,7 +59,7 @@ wrapper.java.additional.11=-Dactivemq.data="%ACTIVEMQ_DATA%"
wrapper.java.additional.12=-Dhawtio.realm=activemq
wrapper.java.additional.13=-Dhawtio.role=admins
wrapper.java.additional.14=-Dhawtio.rolePrincipalClasses=org.apache.activemq.jaas.GroupPrincipal
-wrapper.java.additional.15=-Djava.security.auth.login.config=%ACTIVEMQ_CONF%/login.config
+wrapper.java.additional.15=-Djava.security.auth.login.config="%ACTIVEMQ_CONF%/login.config"
# Uncomment to enable remote jmx
#wrapper.java.additional.n=-Dcom.sun.management.jmxremote.port=1616
[11/13] git commit: Fix for
https://issues.apache.org/jira/browse/AMQ-4825
Posted by ha...@apache.org.
Fix for https://issues.apache.org/jira/browse/AMQ-4825
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/cdbd6225
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/cdbd6225
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/cdbd6225
Branch: refs/heads/activemq-5.9
Commit: cdbd622523732ca7827dce05ae2c44eeb9cb26fa
Parents: bc0d4bb
Author: rajdavies <ra...@gmail.com>
Authored: Tue Dec 10 09:09:21 2013 +0000
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Thu Mar 13 11:07:32 2014 -0400
----------------------------------------------------------------------
.../activemq/broker/jmx/NetworkBridgeView.java | 5 ++
.../broker/jmx/NetworkBridgeViewMBean.java | 2 +
.../network/DemandForwardingBridgeSupport.java | 5 ++
.../apache/activemq/network/NetworkBridge.java | 5 ++
.../activemq/ActiveMQConnectionFactory.java | 60 +++++++++++++++++++-
.../org/apache/activemq/bugs/AMQ4160Test.java | 5 ++
6 files changed, 80 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/cdbd6225/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
index 47f167a..c557754 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
@@ -51,6 +51,11 @@ public class NetworkBridgeView implements NetworkBridgeViewMBean {
return bridge.getRemoteBrokerName();
}
+
+ public String getRemoteBrokerId() {
+ return bridge.getRemoteBrokerId();
+ }
+
public String getLocalBrokerName() {
return bridge.getLocalBrokerName();
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/cdbd6225/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
index b8733d1..639a81a 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
@@ -26,6 +26,8 @@ public interface NetworkBridgeViewMBean extends Service {
String getRemoteBrokerName();
+ String getRemoteBrokerId();
+
String getLocalBrokerName();
long getEnqueueCounter();
http://git-wip-us.apache.org/repos/asf/activemq/blob/cdbd6225/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
index 9fa38a4..1126d22 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
@@ -1499,6 +1499,11 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
}
@Override
+ public String getRemoteBrokerId() {
+ return (remoteBrokerInfo == null || remoteBrokerInfo.getBrokerId() ==null)? null : remoteBrokerInfo.getBrokerId().toString();
+ }
+
+ @Override
public String getLocalBrokerName() {
return localBrokerInfo == null ? null : localBrokerInfo.getBrokerName();
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/cdbd6225/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
index 95d0477..4ed03dc 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
@@ -54,6 +54,11 @@ public interface NetworkBridge extends Service {
String getRemoteBrokerName();
/**
+ * @return the id of the remote broker this bridge is connected to.
+ */
+ String getRemoteBrokerId();
+
+ /**
* @return the network address of the local broker connection.
*/
String getLocalAddress();
http://git-wip-us.apache.org/repos/asf/activemq/blob/cdbd6225/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
----------------------------------------------------------------------
diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
index 58fee6e..8495d84 100755
--- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
+++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
@@ -18,6 +18,8 @@ package org.apache.activemq;
import java.net.URI;
import java.net.URISyntaxException;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
@@ -32,7 +34,6 @@ import javax.jms.QueueConnectionFactory;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.naming.Context;
-
import org.apache.activemq.blob.BlobTransferPolicy;
import org.apache.activemq.broker.region.policy.RedeliveryPolicyMap;
import org.apache.activemq.jndi.JNDIBaseStorable;
@@ -48,6 +49,8 @@ import org.apache.activemq.util.IntrospectionSupport;
import org.apache.activemq.util.JMSExceptionSupport;
import org.apache.activemq.util.URISupport;
import org.apache.activemq.util.URISupport.CompositeData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* A ConnectionFactory is an an Administered object, and is used for creating
@@ -59,8 +62,61 @@ import org.apache.activemq.util.URISupport.CompositeData;
* @see javax.jms.ConnectionFactory
*/
public class ActiveMQConnectionFactory extends JNDIBaseStorable implements ConnectionFactory, QueueConnectionFactory, TopicConnectionFactory, StatsCapable, Cloneable {
+ private static final Logger LOG = LoggerFactory.getLogger(ActiveMQConnectionFactory.class);
+ private static final String DEFAULT_BROKER_HOST;
+ private static final int DEFAULT_BROKER_PORT;
+ static{
+ String host = null;
+ String port = null;
+ try {
+ host = AccessController.doPrivileged(new PrivilegedAction<String>() {
+ @Override
+ public String run() {
+ String result = System.getProperty("org.apache.activemq.AMQ_HOST");
+ result = (result==null||result.isEmpty()) ? System.getProperty("AMQ_HOST","localhost") : result;
+ return result;
+ }
+ });
+ port = AccessController.doPrivileged(new PrivilegedAction<String>() {
+ @Override
+ public String run() {
+ String result = System.getProperty("org.apache.activemq.AMQ_PORT");
+ result = (result==null||result.isEmpty()) ? System.getProperty("AMQ_PORT","61616") : result;
+ return result;
+ }
+ });
+ }catch(Throwable e){
+ LOG.debug("Failed to look up System properties for host and port",e);
+ }
+ host = (host == null || host.isEmpty()) ? "localhost" : host;
+ port = (port == null || port.isEmpty()) ? "61616" : port;
+ DEFAULT_BROKER_HOST = host;
+ DEFAULT_BROKER_PORT = Integer.parseInt(port);
+ }
+
+
+ public static final String DEFAULT_BROKER_BIND_URL;
+
+ static{
+ final String defaultURL = "tcp://" + DEFAULT_BROKER_HOST + ":" + DEFAULT_BROKER_PORT;
+ String bindURL = null;
+
+ try {
+ bindURL = AccessController.doPrivileged(new PrivilegedAction<String>() {
+ @Override
+ public String run() {
+ String result = System.getProperty("org.apache.activemq.BROKER_BIND_URL");
+ result = (result==null||result.isEmpty()) ? System.getProperty("BROKER_BIND_URL",defaultURL) : result;
+ return result;
+ }
+ });
+ }catch(Throwable e){
+ LOG.debug("Failed to look up System properties for host and port",e);
+ }
+ bindURL = (bindURL == null || bindURL.isEmpty()) ? defaultURL : bindURL;
+ DEFAULT_BROKER_BIND_URL = bindURL;
+ }
- public static final String DEFAULT_BROKER_BIND_URL = "tcp://localhost:61616";
public static final String DEFAULT_BROKER_URL = "failover://"+DEFAULT_BROKER_BIND_URL;
public static final String DEFAULT_USER = null;
public static final String DEFAULT_PASSWORD = null;
http://git-wip-us.apache.org/repos/asf/activemq/blob/cdbd6225/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
index 34bff2d..4867f28 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
@@ -303,6 +303,11 @@ public class AMQ4160Test extends JmsMultipleBrokersTestSupport {
}
@Override
+ public String getRemoteBrokerId() {
+ return next.getRemoteBrokerId();
+ }
+
+ @Override
public String getLocalAddress() {
return next.getLocalAddress();
}
[13/13] git commit: Merge for fix for AMQ-4927
Posted by ha...@apache.org.
Merge for fix for AMQ-4927
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/88959d5c
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/88959d5c
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/88959d5c
Branch: refs/heads/activemq-5.9
Commit: 88959d5c653b78df4c629b380f7215129f08228a
Parents: 6d236b0
Author: Hadrian Zbarcea <ha...@apache.org>
Authored: Thu Mar 13 11:20:51 2014 -0400
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Thu Mar 13 11:20:51 2014 -0400
----------------------------------------------------------------------
.../apache/activemq/broker/jmx/ProducerView.java | 4 ++--
.../activemq/broker/jmx/ProducerViewMBean.java | 2 +-
.../activemq/broker/jmx/SubscriptionView.java | 4 ++--
.../activemq/broker/region/AbstractRegion.java | 3 ++-
.../broker/region/AbstractSubscription.java | 19 +++++++++++++------
.../activemq/broker/region/Subscription.java | 14 +++++++-------
.../apache/activemq/command/ProducerInfo.java | 17 +++++++++++++----
.../region/QueueDuplicatesFromStoreTest.java | 12 ++++++++++--
8 files changed, 50 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/88959d5c/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerView.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerView.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerView.java
index 6905c72..e211b75 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerView.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerView.java
@@ -188,12 +188,12 @@ public class ProducerView implements ProducerViewMBean {
@Override
public void resetStatistics() {
if (info != null){
- info.getSentCount().reset();
+ info.resetSentCount();
}
}
@Override
public long getSentCount() {
- return info != null ? info.getSentCount().getCount() :0;
+ return info != null ? info.getSentCount() :0;
}
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/88959d5c/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerViewMBean.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerViewMBean.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerViewMBean.java
index 14c2073..4776283 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerViewMBean.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerViewMBean.java
@@ -102,7 +102,7 @@ public interface ProducerViewMBean {
@MBeanInfo("Resets statistics.")
void resetStatistics();
- @MBeanInfo("Messages consumed")
+ @MBeanInfo("Messages dispatched by Producer")
long getSentCount();
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/88959d5c/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/SubscriptionView.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/SubscriptionView.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/SubscriptionView.java
index 443a266..deefdb4 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/SubscriptionView.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/SubscriptionView.java
@@ -421,12 +421,12 @@ public class SubscriptionView implements SubscriptionViewMBean {
@Override
public void resetStatistics() {
if (subscription != null){
- subscription.getConsumedCount().reset();
+ subscription.resetConsumedCount();
}
}
@Override
public long getConsumedCount() {
- return subscription != null ? subscription.getConsumedCount().getCount() : 0;
+ return subscription != null ? subscription.getConsumedCount() : 0;
}
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/88959d5c/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
index 16deed4..efa02cb 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
@@ -392,8 +392,9 @@ public abstract class AbstractRegion implements Region {
}
producerExchange.getRegionDestination().send(producerExchange, messageSend);
+
if (producerExchange.getProducerState() != null && producerExchange.getProducerState().getInfo() != null){
- producerExchange.getProducerState().getInfo().getSentCount().increment();
+ producerExchange.getProducerState().getInfo().incrementSentCount();
}
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/88959d5c/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractSubscription.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractSubscription.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractSubscription.java
index b2ff01c..3a2e2ee 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractSubscription.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractSubscription.java
@@ -20,11 +20,11 @@ import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicLong;
import javax.jms.InvalidSelectorException;
import javax.jms.JMSException;
import javax.management.ObjectName;
-
import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.command.ActiveMQDestination;
@@ -36,7 +36,6 @@ import org.apache.activemq.filter.DestinationFilter;
import org.apache.activemq.filter.LogicExpression;
import org.apache.activemq.filter.MessageEvaluationContext;
import org.apache.activemq.filter.NoLocalExpression;
-import org.apache.activemq.management.CountStatisticImpl;
import org.apache.activemq.selector.SelectorParser;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,7 +53,7 @@ public abstract class AbstractSubscription implements Subscription {
private int cursorMemoryHighWaterMark = 70;
private boolean slowConsumer;
private long lastAckTime;
- private CountStatisticImpl consumedCount = new CountStatisticImpl("consumed","The number of messages consumed");
+ private AtomicLong consumedCount = new AtomicLong();
public AbstractSubscription(Broker broker,ConnectionContext context, ConsumerInfo info) throws InvalidSelectorException {
this.broker = broker;
@@ -90,7 +89,7 @@ public abstract class AbstractSubscription implements Subscription {
@Override
public synchronized void acknowledge(final ConnectionContext context, final MessageAck ack) throws Exception {
this.lastAckTime = System.currentTimeMillis();
- this.consumedCount.increment();
+ this.consumedCount.incrementAndGet();
}
@Override
@@ -280,7 +279,15 @@ public abstract class AbstractSubscription implements Subscription {
this.lastAckTime = value;
}
- public CountStatisticImpl getConsumedCount(){
- return consumedCount;
+ public long getConsumedCount(){
+ return consumedCount.get();
+ }
+
+ public void incrementConsumedCount(){
+ consumedCount.incrementAndGet();
+ }
+
+ public void resetConsumedCount(){
+ consumedCount.set(0);
}
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/88959d5c/activemq-broker/src/main/java/org/apache/activemq/broker/region/Subscription.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Subscription.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Subscription.java
index b79b37e..a2c4502 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Subscription.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Subscription.java
@@ -21,7 +21,6 @@ import java.util.List;
import javax.jms.InvalidSelectorException;
import javax.management.ObjectName;
-
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ConsumerInfo;
@@ -30,7 +29,6 @@ import org.apache.activemq.command.MessageDispatchNotification;
import org.apache.activemq.command.MessagePull;
import org.apache.activemq.command.Response;
import org.apache.activemq.filter.MessageEvaluationContext;
-import org.apache.activemq.management.CountStatisticImpl;
/**
*
@@ -48,7 +46,6 @@ public interface Subscription extends SubscriptionRecovery {
/**
* Used when client acknowledge receipt of dispatched message.
- * @param node
* @throws IOException
* @throws Exception
*/
@@ -70,7 +67,7 @@ public interface Subscription extends SubscriptionRecovery {
/**
* Is the subscription interested in messages in the destination?
- * @param context
+ * @param destination
* @return
*/
boolean matches(ActiveMQDestination destination);
@@ -93,7 +90,6 @@ public interface Subscription extends SubscriptionRecovery {
/**
* The ConsumerInfo object that created the subscription.
- * @param destination
*/
ConsumerInfo getConsumerInfo();
@@ -200,7 +196,7 @@ public interface Subscription extends SubscriptionRecovery {
/**
* Informs the Broker if the subscription needs to intervention to recover it's state
* e.g. DurableTopicSubscriber may do
- * @see org.apache.activemq.region.cursors.PendingMessageCursor
+ * @see org.apache.activemq.broker.region.cursors.PendingMessageCursor
* @return true if recovery required
*/
boolean isRecoveryRequired();
@@ -235,6 +231,10 @@ public interface Subscription extends SubscriptionRecovery {
*/
long getTimeOfLastMessageAck();
- CountStatisticImpl getConsumedCount();
+ long getConsumedCount();
+
+ void incrementConsumedCount();
+
+ void resetConsumedCount();
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/88959d5c/activemq-client/src/main/java/org/apache/activemq/command/ProducerInfo.java
----------------------------------------------------------------------
diff --git a/activemq-client/src/main/java/org/apache/activemq/command/ProducerInfo.java b/activemq-client/src/main/java/org/apache/activemq/command/ProducerInfo.java
index 05ef3a4..7189347 100755
--- a/activemq-client/src/main/java/org/apache/activemq/command/ProducerInfo.java
+++ b/activemq-client/src/main/java/org/apache/activemq/command/ProducerInfo.java
@@ -16,7 +16,8 @@
*/
package org.apache.activemq.command;
-import org.apache.activemq.management.CountStatisticImpl;
+import java.util.concurrent.atomic.AtomicLong;
+
import org.apache.activemq.state.CommandVisitor;
/**
@@ -33,7 +34,7 @@ public class ProducerInfo extends BaseCommand {
protected BrokerId[] brokerPath;
protected boolean dispatchAsync;
protected int windowSize;
- protected CountStatisticImpl sentCount = new CountStatisticImpl("sentCount","number of messages sent to a broker");
+ protected AtomicLong sentCount = new AtomicLong();
public ProducerInfo() {
}
@@ -137,8 +138,16 @@ public class ProducerInfo extends BaseCommand {
this.windowSize = windowSize;
}
- public CountStatisticImpl getSentCount(){
- return sentCount;
+ public long getSentCount(){
+ return sentCount.get();
+ }
+
+ public void incrementSentCount(){
+ sentCount.incrementAndGet();
+ }
+
+ public void resetSentCount(){
+ sentCount.set(0);
}
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/88959d5c/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/QueueDuplicatesFromStoreTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/QueueDuplicatesFromStoreTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/QueueDuplicatesFromStoreTest.java
index 9da839d..e9c6664 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/QueueDuplicatesFromStoreTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/QueueDuplicatesFromStoreTest.java
@@ -342,8 +342,16 @@ public class QueueDuplicatesFromStoreTest extends TestCase {
}
@Override
- public CountStatisticImpl getConsumedCount() {
- return null;
+ public long getConsumedCount() {
+ return 0;
+ }
+
+ public void incrementConsumedCount(){
+
+ }
+
+ public void resetConsumedCount(){
+
}
};
[05/13] git commit: Fixes
https://issues.apache.org/jira/browse/AMQ-4923: Replicated LevelDB: Loss of
broker Quorum fails to fully stop the master
Posted by ha...@apache.org.
Fixes https://issues.apache.org/jira/browse/AMQ-4923: Replicated LevelDB: Loss of broker Quorum fails to fully stop the master
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/bc2f708c
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/bc2f708c
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/bc2f708c
Branch: refs/heads/activemq-5.9
Commit: bc2f708c3c144a775ca25762debf9baa0c7891c9
Parents: d26545c
Author: Hiram Chirino <hi...@hiramchirino.com>
Authored: Thu Dec 5 13:27:59 2013 -0500
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Thu Mar 13 11:05:55 2014 -0400
----------------------------------------------------------------------
.../apache/activemq/broker/BrokerService.java | 4 +
.../org/apache/activemq/leveldb/DBManager.scala | 11 +-
.../test/ReplicatedLevelDBBrokerTest.java | 191 ++++++++++++++++++-
3 files changed, 199 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/bc2f708c/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java b/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
index 575e313..fe6f59b 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
@@ -878,6 +878,10 @@ public class BrokerService implements Service {
}
}
+ public boolean isStopped() {
+ return stopped.get();
+ }
+
/**
* A helper method to block the caller thread until the broker has fully started
* @return boolean true if wait succeeded false if broker was not started or was stopped
http://git-wip-us.apache.org/repos/asf/activemq/blob/bc2f708c/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
----------------------------------------------------------------------
diff --git a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
index 722d932..cfcce78 100644
--- a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
+++ b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
@@ -35,6 +35,7 @@ import util.TimeMetric
import scala.Some
import org.apache.activemq.ActiveMQMessageAuditNoSync
import org.fusesource.hawtdispatch
+import org.apache.activemq.broker.SuppressReplyException
case class EntryLocator(qid:Long, seq:Long)
case class DataLocator(store:LevelDBStore, pos:Long, len:Int) {
@@ -569,9 +570,6 @@ class DBManager(val parent:LevelDBStore) {
def drainFlushes:Unit = {
dispatchQueue.assertExecuting()
- if( !started ) {
- return
- }
// Some UOWs may have been canceled.
import collection.JavaConversions._
@@ -590,7 +588,12 @@ class DBManager(val parent:LevelDBStore) {
assert(action!=null)
}
}
- Some(uow)
+ if( !started ) {
+ uow.onCompleted(new SuppressReplyException("Store stopped"))
+ None
+ } else {
+ Some(uow)
+ }
}
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/bc2f708c/activemq-leveldb-store/src/test/java/org/apache/activemq/leveldb/test/ReplicatedLevelDBBrokerTest.java
----------------------------------------------------------------------
diff --git a/activemq-leveldb-store/src/test/java/org/apache/activemq/leveldb/test/ReplicatedLevelDBBrokerTest.java b/activemq-leveldb-store/src/test/java/org/apache/activemq/leveldb/test/ReplicatedLevelDBBrokerTest.java
index a8e743f..8910981 100644
--- a/activemq-leveldb-store/src/test/java/org/apache/activemq/leveldb/test/ReplicatedLevelDBBrokerTest.java
+++ b/activemq-leveldb-store/src/test/java/org/apache/activemq/leveldb/test/ReplicatedLevelDBBrokerTest.java
@@ -22,6 +22,7 @@ import org.apache.activemq.broker.TransportConnector;
import org.apache.activemq.leveldb.replicated.ElectingLevelDBStore;
import org.junit.After;
import org.junit.Before;
+import org.junit.Ignore;
import org.junit.Test;
import javax.jms.*;
@@ -30,10 +31,17 @@ import javax.management.openmbean.CompositeData;
import java.io.File;
import java.io.IOException;
import java.lang.management.ManagementFactory;
+import java.net.ServerSocket;
+import java.net.URI;
+import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Enumeration;
+import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
import org.apache.commons.io.FileUtils;
import static org.junit.Assert.*;
@@ -82,6 +90,165 @@ public class ReplicatedLevelDBBrokerTest extends ZooKeeperTestSupport {
}
}
+ public interface Client{
+ public void execute(Connection connection) throws Exception;
+ }
+
+ protected Thread startFailoverClient(String name, final Client client) throws IOException, URISyntaxException {
+ String url = "failover://(tcp://localhost:"+port+")?maxReconnectDelay=500&nested.wireFormat.maxInactivityDuration=1000";
+ final ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);
+ Thread rc = new Thread(name) {
+ @Override
+ public void run() {
+ Connection connection = null;
+ try {
+ connection = factory.createConnection();
+ client.execute(connection);
+ } catch (Throwable e) {
+ e.printStackTrace();
+ } finally {
+ try {
+ connection.close();
+ } catch (JMSException e) {
+ }
+ }
+ }
+ };
+ rc.start();
+ return rc;
+ }
+
+ @Test
+ @Ignore
+ public void testReplicationQuorumLoss() throws Throwable {
+
+ System.out.println("======================================");
+ System.out.println(" Start 2 ActiveMQ nodes.");
+ System.out.println("======================================");
+ startBrokerAsync(createBrokerNode("node-1", port));
+ startBrokerAsync(createBrokerNode("node-2", port));
+ BrokerService master = waitForNextMaster();
+ System.out.println("======================================");
+ System.out.println(" Start the producer and consumer");
+ System.out.println("======================================");
+
+ final AtomicBoolean stopClients = new AtomicBoolean(false);
+ final ArrayBlockingQueue<String> errors = new ArrayBlockingQueue<String>(100);
+ final AtomicLong receivedCounter = new AtomicLong();
+ final AtomicLong sentCounter = new AtomicLong();
+ Thread producer = startFailoverClient("producer", new Client() {
+ @Override
+ public void execute(Connection connection) throws Exception {
+ Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(session.createQueue("test"));
+ long actual = 0;
+ while(!stopClients.get()) {
+ TextMessage msg = session.createTextMessage("Hello World");
+ msg.setLongProperty("id", actual++);
+ producer.send(msg);
+ sentCounter.incrementAndGet();
+ }
+ }
+ });
+
+ Thread consumer = startFailoverClient("consumer", new Client() {
+ @Override
+ public void execute(Connection connection) throws Exception {
+ connection.start();
+ Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+ MessageConsumer consumer = session.createConsumer(session.createQueue("test"));
+ long expected = 0;
+ while(!stopClients.get()) {
+ Message msg = consumer.receive(200);
+ if( msg!=null ) {
+ long actual = msg.getLongProperty("id");
+ if( actual != expected ) {
+ errors.offer("Received got unexpected msg id: "+actual+", expected: "+expected);
+ }
+ msg.acknowledge();
+ expected = actual+1;
+ receivedCounter.incrementAndGet();
+ }
+ }
+ }
+ });
+
+ try {
+ assertCounterMakesProgress(sentCounter, 10, TimeUnit.SECONDS);
+ assertCounterMakesProgress(receivedCounter, 5, TimeUnit.SECONDS);
+ assertNull(errors.poll());
+
+ System.out.println("======================================");
+ System.out.println(" Master should stop once the quorum is lost.");
+ System.out.println("======================================");
+ ArrayList<BrokerService> stopped = stopSlaves();// stopping the slaves should kill the quorum.
+ assertStopsWithin(master, 10, TimeUnit.SECONDS);
+ assertNull(errors.poll()); // clients should not see an error since they are failover clients.
+ stopped.add(master);
+
+ System.out.println("======================================");
+ System.out.println(" Restart the slave. Clients should make progress again..");
+ System.out.println("======================================");
+ startBrokersAsync(createBrokerNodes(stopped));
+ assertCounterMakesProgress(sentCounter, 10, TimeUnit.SECONDS);
+ assertCounterMakesProgress(receivedCounter, 5, TimeUnit.SECONDS);
+ assertNull(errors.poll());
+ } catch (Throwable e) {
+ e.printStackTrace();
+ throw e;
+ } finally {
+ // Wait for the clients to stop..
+ stopClients.set(true);
+ producer.join();
+ consumer.join();
+ }
+ }
+
+ protected void startBrokersAsync(ArrayList<BrokerService> brokers) {
+ for (BrokerService broker : brokers) {
+ startBrokerAsync(broker);
+ }
+ }
+
+ protected ArrayList<BrokerService> createBrokerNodes(ArrayList<BrokerService> brokers) throws Exception {
+ ArrayList<BrokerService> rc = new ArrayList<BrokerService>();
+ for (BrokerService b : brokers) {
+ rc.add(createBrokerNode(b.getBrokerName(), connectPort(b)));
+ }
+ return rc;
+ }
+
+ protected ArrayList<BrokerService> stopSlaves() throws Exception {
+ ArrayList<BrokerService> rc = new ArrayList<BrokerService>();
+ for (BrokerService broker : brokers) {
+ if( broker.isSlave() ) {
+ System.out.println("Stopping slave: "+broker.getBrokerName());
+ broker.stop();
+ broker.waitUntilStopped();
+ rc.add(broker);
+ }
+ }
+ brokers.removeAll(rc);
+ return rc;
+ }
+
+ protected void assertStopsWithin(final BrokerService master, int timeout, TimeUnit unit) throws InterruptedException {
+ within(timeout, unit, new Task(){
+ @Override
+ public void run() throws Exception {
+ assertTrue(master.isStopped());
+ }
+ });
+ }
+
+ protected void assertCounterMakesProgress(final AtomicLong counter, int timeout, TimeUnit unit) throws InterruptedException {
+ final long initial = counter.get();
+ within(timeout, unit, new Task(){
+ public void run() throws Exception {
+ assertTrue(initial < counter.get());
+ }
+ });
+ }
public void testAMQ4837(boolean jmx) throws Throwable {
@@ -205,8 +372,7 @@ public class ReplicatedLevelDBBrokerTest extends ZooKeeperTestSupport {
private ArrayList<String> browseMessagesViaJMS(BrokerService brokerService) throws Exception {
ArrayList<String> rc = new ArrayList<String>();
- TransportConnector connector = brokerService.getTransportConnectors().get(0);
- ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(connector.getConnectUri());
+ ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:"+ connectPort(brokerService));
Connection connection = factory.createConnection();
try {
connection.start();
@@ -223,6 +389,19 @@ public class ReplicatedLevelDBBrokerTest extends ZooKeeperTestSupport {
return rc;
}
+ private int connectPort(BrokerService brokerService) throws IOException, URISyntaxException {
+ TransportConnector connector = brokerService.getTransportConnectors().get(0);
+ return connector.getConnectUri().getPort();
+ }
+
+ int port;
+ @Before
+ public void findFreePort() throws Exception {
+ ServerSocket socket = new ServerSocket(0);
+ port = socket.getLocalPort();
+ socket.close();
+ }
+
@After
public void stopBrokers() throws Exception {
for (BrokerService broker : brokers) {
@@ -235,12 +414,18 @@ public class ReplicatedLevelDBBrokerTest extends ZooKeeperTestSupport {
}
private BrokerService createBrokerNode(String id) throws Exception {
+ return createBrokerNode(id, 0);
+ }
+
+ private BrokerService createBrokerNode(String id, int port) throws Exception {
BrokerService bs = new BrokerService();
bs.getManagementContext().setCreateConnector(false);
brokers.add(bs);
bs.setBrokerName(id);
bs.setPersistenceAdapter(createStoreNode(id));
- bs.addConnector("tcp://0.0.0.0:0");
+ TransportConnector connector = new TransportConnector();
+ connector.setUri(new URI("tcp://0.0.0.0:" + port));
+ bs.addConnector(connector);
return bs;
}
[07/13] git commit: https://issues.apache.org/jira/browse/AMQ-4887
Posted by ha...@apache.org.
https://issues.apache.org/jira/browse/AMQ-4887
Update test case to show other broken scenarios.
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/f713c827
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/f713c827
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/f713c827
Branch: refs/heads/activemq-5.9
Commit: f713c827bfb83ab6a98edb4e9946847f63f1548e
Parents: 7f68893
Author: Timothy Bish <ta...@gmai.com>
Authored: Thu Dec 5 15:52:05 2013 -0500
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Thu Mar 13 11:06:28 2014 -0400
----------------------------------------------------------------------
.../org/apache/activemq/bugs/AMQ4887Test.java | 124 ++++++++++++++++---
1 file changed, 104 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/f713c827/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4887Test.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4887Test.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4887Test.java
index f29f605..cf33ece 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4887Test.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4887Test.java
@@ -16,48 +16,65 @@
*/
package org.apache.activemq.bugs;
-import org.apache.activemq.ActiveMQConnection;
-import org.apache.activemq.ActiveMQConnectionFactory;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TestName;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
import javax.jms.BytesMessage;
+import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
+import javax.jms.StreamMessage;
-import static org.junit.Assert.*;
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class AMQ4887Test {
private static final transient Logger LOG = LoggerFactory.getLogger(AMQ4887Test.class);
private static final Integer ITERATIONS = 10;
+
@Rule
public TestName name = new TestName();
@Test
- public void testSetPropertyBeforeCopy() throws Exception {
+ public void testBytesMessageSetPropertyBeforeCopy() throws Exception {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");
ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();
connection.start();
+ doTestBytesMessageSetPropertyBeforeCopy(connection);
+ }
+
+ @Test
+ public void testBytesMessageSetPropertyBeforeCopyCompressed() throws Exception {
+ ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");
+ connectionFactory.setUseCompression(true);
+ ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();
+ connection.start();
+ doTestBytesMessageSetPropertyBeforeCopy(connection);
+ }
+
+ public void doTestBytesMessageSetPropertyBeforeCopy(Connection connection) throws Exception {
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue(name.toString());
MessageConsumer consumer = session.createConsumer(destination);
MessageProducer producer = session.createProducer(destination);
- byte[] messageContent = "bytes message".getBytes();
- BytesMessage bytesMessage = session.createBytesMessage();
- bytesMessage.writeBytes(messageContent);
+ BytesMessage message = session.createBytesMessage();
for (int i=0; i < ITERATIONS; i++) {
+
long sendTime = System.currentTimeMillis();
- bytesMessage.setLongProperty("sendTime", sendTime);
- producer.send(bytesMessage);
+ message.setLongProperty("sendTime", sendTime);
+ producer.send(message);
LOG.debug("Receiving message " + i);
Message receivedMessage = consumer.receive(5000);
@@ -65,16 +82,83 @@ public class AMQ4887Test {
assertTrue("On message " + i, receivedMessage instanceof BytesMessage);
BytesMessage receivedBytesMessage = (BytesMessage) receivedMessage;
- byte[] receivedBytes = new byte[(int) receivedBytesMessage.getBodyLength()];
- receivedBytesMessage.readBytes(receivedBytes);
- LOG.debug("Message " + i + " content [" + new String(receivedBytes) + "]");
- assertEquals("On message " + i, messageContent.length, receivedBytes.length);
- assertArrayEquals("On message " + i, messageContent, receivedBytes);
+
+ int numElements = 0;
+ try {
+ while (true) {
+ receivedBytesMessage.readBoolean();
+ numElements++;
+ }
+ } catch (Exception ex) {
+ }
+
+ LOG.info("Iteration [{}]: Received Message contained {} boolean values.", i, numElements);
+ assertEquals(i, numElements);
long receivedSendTime = receivedBytesMessage.getLongProperty("sendTime");
assertEquals("On message " + i, receivedSendTime, sendTime);
- Thread.sleep(10);
+ // Add a new bool value on each iteration.
+ message.writeBoolean(true);
+ }
+ }
+
+ @Test
+ public void testStreamMessageSetPropertyBeforeCopy() throws Exception {
+ ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");
+ ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();
+ connection.start();
+ doTestStreamMessageSetPropertyBeforeCopy(connection);
+ }
+
+ @Test
+ public void testStreamMessageSetPropertyBeforeCopyCompressed() throws Exception {
+ ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");
+ connectionFactory.setUseCompression(true);
+ ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();
+ connection.start();
+ doTestStreamMessageSetPropertyBeforeCopy(connection);
+ }
+
+ public void doTestStreamMessageSetPropertyBeforeCopy(Connection connection) throws Exception {
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Destination destination = session.createQueue(name.toString());
+ MessageConsumer consumer = session.createConsumer(destination);
+ MessageProducer producer = session.createProducer(destination);
+
+ StreamMessage message = session.createStreamMessage();
+
+ for (int i=0; i < ITERATIONS; i++) {
+
+ long sendTime = System.currentTimeMillis();
+ message.setLongProperty("sendTime", sendTime);
+ producer.send(message);
+
+ LOG.debug("Receiving message " + i);
+ Message receivedMessage = consumer.receive(5000);
+ assertNotNull("On message " + i, receivedMessage);
+ assertTrue("On message " + i, receivedMessage instanceof StreamMessage);
+
+ StreamMessage receivedStreamMessage = (StreamMessage) receivedMessage;
+
+ int numElements = 0;
+ try {
+ while (true) {
+ receivedStreamMessage.readBoolean();
+ numElements++;
+ }
+ } catch (Exception ex) {
+ }
+
+ LOG.info("Iteration [{}]: Received Message contained {} boolean values.", i, numElements);
+ assertEquals(i, numElements);
+
+ long receivedSendTime = receivedStreamMessage.getLongProperty("sendTime");
+ assertEquals("On message " + i, receivedSendTime, sendTime);
+
+ // Add a new bool value on each iteration.
+ message.writeBoolean(true);
}
}
[10/13] git commit: https://issues.apache.org/jira/browse/AMQ-1063 -
revert delayed change to inTx flag,
now cleared before exception is thrown by getConnection,
this ensures jms transactions can use autocommit=false
Posted by ha...@apache.org.
https://issues.apache.org/jira/browse/AMQ-1063 - revert delayed change to inTx flag, now cleared before exception is thrown by getConnection, this ensures jms transactions can use autocommit=false
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/bc0d4bb5
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/bc0d4bb5
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/bc0d4bb5
Branch: refs/heads/activemq-5.9
Commit: bc0d4bb5752bbc3695ba4f73a698a0480b8d6619
Parents: 015c6cc
Author: gtully <ga...@gmail.com>
Authored: Mon Dec 9 17:01:56 2013 +0000
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Thu Mar 13 11:07:17 2014 -0400
----------------------------------------------------------------------
.../org/apache/activemq/store/jdbc/TransactionContext.java | 7 ++++---
1 file changed, 4 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/bc0d4bb5/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/TransactionContext.java
----------------------------------------------------------------------
diff --git a/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/TransactionContext.java b/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/TransactionContext.java
index 66163b6..6a933b0 100755
--- a/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/TransactionContext.java
+++ b/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/TransactionContext.java
@@ -38,7 +38,7 @@ public class TransactionContext {
private final DataSource dataSource;
private final JDBCPersistenceAdapter persistenceAdapter;
private Connection connection;
- private volatile boolean inTx;
+ private boolean inTx;
private PreparedStatement addMessageStatement;
private PreparedStatement removedMessageStatement;
private PreparedStatement updateLastAckStatement;
@@ -63,6 +63,8 @@ public class TransactionContext {
}
} catch (SQLException e) {
JDBCPersistenceAdapter.log("Could not get JDBC connection: ", e);
+ inTx = false;
+ close();
IOException ioe = IOExceptionSupport.create(e);
persistenceAdapter.getBrokerService().handleIOException(ioe);
throw ioe;
@@ -160,9 +162,8 @@ public class TransactionContext {
if (inTx) {
throw new IOException("Already started.");
}
- connection = getConnection();
- // only mark in tx if we could get a connection
inTx = true;
+ connection = getConnection();
}
public void commit() throws IOException {
[09/13] git commit: Fix for
https://issues.apache.org/jira/browse/AMQ-4912
Posted by ha...@apache.org.
Fix for https://issues.apache.org/jira/browse/AMQ-4912
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/015c6cce
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/015c6cce
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/015c6cce
Branch: refs/heads/activemq-5.9
Commit: 015c6cce7341d03cecfd3e1f1a52c86a9f3b78e9
Parents: 4a80e95
Author: rajdavies <ra...@gmail.com>
Authored: Mon Dec 9 16:59:26 2013 +0000
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Thu Mar 13 11:06:57 2014 -0400
----------------------------------------------------------------------
.../main/java/org/apache/activemq/plugin/StatisticsBroker.java | 5 +++++
1 file changed, 5 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/015c6cce/activemq-broker/src/main/java/org/apache/activemq/plugin/StatisticsBroker.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/plugin/StatisticsBroker.java b/activemq-broker/src/main/java/org/apache/activemq/plugin/StatisticsBroker.java
index 9679274..c1949c5 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/plugin/StatisticsBroker.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/plugin/StatisticsBroker.java
@@ -104,6 +104,8 @@ public class StatisticsBroker extends BrokerFilter {
DestinationStatistics stats = dest.getDestinationStatistics();
if (stats != null) {
ActiveMQMapMessage statsMessage = new ActiveMQMapMessage();
+ statsMessage.setString("brokerName", regionBroker.getBrokerName());
+ statsMessage.setString("brokerId", regionBroker.getBrokerId().toString());
statsMessage.setString("destinationName", dest.getActiveMQDestination().toString());
statsMessage.setLong("size", stats.getMessages().getCount());
statsMessage.setLong("enqueueCount", stats.getEnqueues().getCount());
@@ -223,7 +225,10 @@ public class StatisticsBroker extends BrokerFilter {
}
protected ActiveMQMapMessage prepareSubscriptionMessage(SubscriptionViewMBean subscriber) throws JMSException {
+ Broker regionBroker = getBrokerService().getRegionBroker();
ActiveMQMapMessage statsMessage = new ActiveMQMapMessage();
+ statsMessage.setString("brokerName", regionBroker.getBrokerName());
+ statsMessage.setString("brokerId", regionBroker.getBrokerId().toString());
statsMessage.setString("destinationName", subscriber.getDestinationName());
statsMessage.setString("clientId", subscriber.getClientId());
statsMessage.setString("connectionId", subscriber.getConnectionId());
[03/13] git commit: Implementing a dirty stop method for the leveldb
store that might come in handy for testing.
Posted by ha...@apache.org.
Implementing a dirty stop method for the leveldb store that might come in handy for testing.
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/6a7e640d
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/6a7e640d
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/6a7e640d
Branch: refs/heads/activemq-5.9
Commit: 6a7e640d33c49476004a52d7218b136c7d6ae78a
Parents: af09bdc
Author: Hiram Chirino <hi...@hiramchirino.com>
Authored: Thu Dec 5 09:23:26 2013 -0500
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Thu Mar 13 11:04:52 2014 -0400
----------------------------------------------------------------------
.../scala/org/apache/activemq/leveldb/LevelDBClient.scala | 10 +++++++++-
.../leveldb/replicated/ElectingLevelDBStore.scala | 1 +
2 files changed, 10 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/6a7e640d/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala
----------------------------------------------------------------------
diff --git a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala
index 7872ff8..44f0d44 100755
--- a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala
+++ b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala
@@ -971,7 +971,15 @@ class LevelDBClient(store: LevelDBStore) {
}
}
- def stop() = this.synchronized {
+ def dirty_stop = this.synchronized {
+ def ingorefailure(func: =>Unit) = try { func } catch { case e:Throwable=> }
+ ingorefailure(index.close)
+ ingorefailure(log.close)
+ ingorefailure(plist.close)
+ ingorefailure(might_fail(throw new IOException("non-clean close")))
+ }
+
+ def stop():Unit = this.synchronized {
if( writeExecutor!=null ) {
writeExecutor.shutdown
writeExecutor.awaitTermination(60, TimeUnit.SECONDS)
http://git-wip-us.apache.org/repos/asf/activemq/blob/6a7e640d/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/ElectingLevelDBStore.scala
----------------------------------------------------------------------
diff --git a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/ElectingLevelDBStore.scala b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/ElectingLevelDBStore.scala
index 90e80c0..55e0419 100644
--- a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/ElectingLevelDBStore.scala
+++ b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/ElectingLevelDBStore.scala
@@ -243,6 +243,7 @@ class ElectingLevelDBStore extends ProxyLevelDBStore {
master_stopped.set(true)
position = master.wal_append_position
stopped_latch.countDown()
+ master = null
func
})
master.blocking_executor.execute(^{