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 2018/10/09 11:02:09 UTC
[1/2] activemq git commit: KahaDB Recovery can experience a dangling
transaction when prepare and commit occur on different pagefiles.
Repository: activemq
Updated Branches:
refs/heads/master ac1e709dc -> 57c793953
KahaDB Recovery can experience a dangling transaction when prepare and commit occur on different pagefiles.
Signed-off-by: gtully <ga...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/99db9ef3
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/99db9ef3
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/99db9ef3
Branch: refs/heads/master
Commit: 99db9ef301c8e8a7156d5c5a71875958f412792a
Parents: ac1e709
Author: jgoodyear <ja...@gmail.com>
Authored: Fri Oct 5 14:04:21 2018 -0230
Committer: gtully <ga...@gmail.com>
Committed: Tue Oct 9 10:19:35 2018 +0100
----------------------------------------------------------------------
.../activemq/store/kahadb/MessageDatabase.java | 11 +
.../org/apache/activemq/bugs/AMQ7067Test.java | 360 +++++++++++++++++++
.../apache/activemq/bugs/amq7067/activemq.xml | 111 ++++++
3 files changed, 482 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/99db9ef3/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 0e5c237..1c974cc 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
@@ -1408,6 +1408,9 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
} finally {
indexLock.writeLock().unlock();
}
+ for (Operation op: inflightTx) {
+ recordAckMessageReferenceLocation(location, op.getLocation());
+ }
}
@SuppressWarnings("rawtypes")
@@ -1417,6 +1420,9 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
List<Operation> tx = inflightTransactions.remove(key);
if (tx != null) {
preparedTransactions.put(key, tx);
+ for (Operation op: tx) {
+ recordAckMessageReferenceLocation(location, op.getLocation());
+ }
}
}
}
@@ -1431,6 +1437,11 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
updates = preparedTransactions.remove(key);
}
}
+ if (updates != null) {
+ for(Operation op : updates) {
+ recordAckMessageReferenceLocation(location, op.getLocation());
+ }
+ }
}
protected void process(KahaRewrittenDataFileCommand command, Location location) throws IOException {
http://git-wip-us.apache.org/repos/asf/activemq/blob/99db9ef3/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ7067Test.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ7067Test.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ7067Test.java
new file mode 100644
index 0000000..c1f34d0
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ7067Test.java
@@ -0,0 +1,360 @@
+package org.apache.activemq.bugs;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.ActiveMQXAConnection;
+import org.apache.activemq.ActiveMQXAConnectionFactory;
+import org.apache.activemq.broker.BrokerFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.jmx.BrokerMBeanSupport;
+import org.apache.activemq.broker.jmx.BrokerViewMBean;
+import org.apache.activemq.broker.jmx.DestinationViewMBean;
+import org.apache.activemq.broker.jmx.PersistenceAdapterViewMBean;
+import org.apache.activemq.broker.jmx.QueueViewMBean;
+import org.apache.activemq.broker.jmx.RecoveredXATransactionViewMBean;
+import org.apache.activemq.command.TransactionId;
+import org.apache.activemq.command.XATransactionId;
+import org.apache.activemq.util.JMXSupport;
+import org.apache.commons.lang.StringUtils;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import javax.jms.*;
+import javax.management.MalformedObjectNameException;
+import javax.management.ObjectName;
+import javax.transaction.xa.XAException;
+import javax.transaction.xa.XAResource;
+import javax.transaction.xa.Xid;
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.util.Random;
+
+import static javax.transaction.xa.XAResource.*;
+import static org.junit.Assert.assertEquals;
+
+public class AMQ7067Test {
+
+ protected static Random r = new Random();
+ final static String WIRE_LEVEL_ENDPOINT = "tcp://localhost:61616";
+ protected BrokerService broker;
+ protected ActiveMQXAConnection connection;
+ protected XASession xaSession;
+ protected XAResource xaRes;
+ private final String xbean = "xbean:";
+ private final String confBase = "src/test/resources/org/apache/activemq/bugs/amq7067";
+
+
+ private static final ActiveMQXAConnectionFactory ACTIVE_MQ_CONNECTION_FACTORY;
+ private static final ActiveMQConnectionFactory ACTIVE_MQ_NON_XA_CONNECTION_FACTORY;
+
+ static {
+ ACTIVE_MQ_CONNECTION_FACTORY = new ActiveMQXAConnectionFactory(WIRE_LEVEL_ENDPOINT);
+ ACTIVE_MQ_NON_XA_CONNECTION_FACTORY = new ActiveMQConnectionFactory(WIRE_LEVEL_ENDPOINT);
+ }
+
+ @Before
+ public void setup() throws Exception {
+ deleteData(new File("target/data"));
+ createBroker();
+ }
+
+ @After
+ public void shutdown() throws Exception {
+ broker.stop();
+ }
+
+ public void setupXAConnection() throws Exception {
+ connection = (ActiveMQXAConnection) ACTIVE_MQ_CONNECTION_FACTORY.createXAConnection();
+ connection.start();
+ xaSession = connection.createXASession();
+ xaRes = xaSession.getXAResource();
+ }
+
+ private void createBroker() throws Exception {
+ broker = new BrokerService();
+ broker = BrokerFactory.createBroker(xbean + confBase + "/activemq.xml");
+ broker.start();
+ }
+
+
+ @Test
+ public void testAMQ7067XAcommit() throws Exception {
+
+ PersistenceAdapterViewMBean kahadbView = getProxyToPersistenceAdapter(broker.getPersistenceAdapter().toString());
+ setupXAConnection();
+
+ Queue holdKahaDb = xaSession.createQueue("holdKahaDb");
+ createDanglingTransaction(xaRes, xaSession, holdKahaDb);
+
+ MessageProducer holdKahaDbProducer = xaSession.createProducer(holdKahaDb);
+
+ XATransactionId txid = createXATransaction();
+ System.out.println("****** create new txid = " + txid);
+ xaRes.start(txid, TMNOFLAGS);
+
+ TextMessage helloMessage = xaSession.createTextMessage(StringUtils.repeat("a", 10));
+ holdKahaDbProducer.send(helloMessage);
+ xaRes.end(txid, TMSUCCESS);
+ xaRes.prepare(txid);
+
+ Queue queue = xaSession.createQueue("test");
+
+ produce(xaRes, xaSession, queue, 100, 512 * 1024);
+ xaRes.commit(txid, false);
+ produce(xaRes, xaSession, queue, 100, 512 * 1024);
+
+ ((org.apache.activemq.broker.region.Queue) broker.getRegionBroker().getDestinationMap().get(queue)).purge();
+
+ Xid[] xids = xaRes.recover(TMSTARTRSCAN);
+
+ //Should be 1 since we have only 1 prepared
+ assertEquals(1, xids.length);
+ connection.close();
+
+ broker.stop();
+ broker.waitUntilStopped();
+ createBroker();
+
+ setupXAConnection();
+ xids = xaRes.recover(TMSTARTRSCAN);
+
+ // THIS SHOULD NOT FAIL AS THERE SHOUL DBE ONLY 1 TRANSACTION!
+ assertEquals(1, xids.length);
+
+ }
+
+ @Test
+ public void testAMQ7067XArollback() throws Exception {
+
+ PersistenceAdapterViewMBean kahadbView = getProxyToPersistenceAdapter(broker.getPersistenceAdapter().toString());
+ setupXAConnection();
+
+ Queue holdKahaDb = xaSession.createQueue("holdKahaDb");
+ createDanglingTransaction(xaRes, xaSession, holdKahaDb);
+
+ MessageProducer holdKahaDbProducer = xaSession.createProducer(holdKahaDb);
+
+ XATransactionId txid = createXATransaction();
+ System.out.println("****** create new txid = " + txid);
+ xaRes.start(txid, TMNOFLAGS);
+
+ TextMessage helloMessage = xaSession.createTextMessage(StringUtils.repeat("a", 10));
+ holdKahaDbProducer.send(helloMessage);
+ xaRes.end(txid, TMSUCCESS);
+ xaRes.prepare(txid);
+
+ Queue queue = xaSession.createQueue("test");
+
+ produce(xaRes, xaSession, queue, 100, 512 * 1024);
+ xaRes.rollback(txid);
+ produce(xaRes, xaSession, queue, 100, 512 * 1024);
+
+ ((org.apache.activemq.broker.region.Queue) broker.getRegionBroker().getDestinationMap().get(queue)).purge();
+
+ Xid[] xids = xaRes.recover(TMSTARTRSCAN);
+
+ //Should be 1 since we have only 1 prepared
+ assertEquals(1, xids.length);
+ connection.close();
+
+ broker.stop();
+ broker.waitUntilStopped();
+ createBroker();
+
+ setupXAConnection();
+ xids = xaRes.recover(TMSTARTRSCAN);
+
+ // THIS SHOULD NOT FAIL AS THERE SHOULD BE ONLY 1 TRANSACTION!
+ assertEquals(1, xids.length);
+
+ }
+
+ @Test
+ public void testAMQ7067commit() throws Exception {
+ final Connection connection = ACTIVE_MQ_NON_XA_CONNECTION_FACTORY.createConnection();
+ connection.start();
+
+ Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
+ Queue holdKahaDb = session.createQueue("holdKahaDb");
+ MessageProducer holdKahaDbProducer = session.createProducer(holdKahaDb);
+ TextMessage helloMessage = session.createTextMessage(StringUtils.repeat("a", 10));
+ holdKahaDbProducer.send(helloMessage);
+ Queue queue = session.createQueue("test");
+ produce(connection, queue, 100, 512*1024);
+ session.commit();
+ produce(connection, queue, 100, 512*1024);
+
+ System.out.println(String.format("QueueSize %s: %d", holdKahaDb.getQueueName(), getQueueSize(holdKahaDb.getQueueName())));
+ purgeQueue(queue.getQueueName());
+ Thread.sleep(10000);
+
+ curruptIndexFile(getDataDirectory());
+
+
+ while(true) {
+ try {
+ Thread.sleep(10000);
+ System.out.println(String.format("QueueSize %s: %d", holdKahaDb.getQueueName(), getQueueSize(holdKahaDb.getQueueName())));
+ break;
+ } catch (Exception ex) {
+ System.out.println(ex.getMessage());
+ break;
+ }
+ }
+
+ connection.close();
+
+ // THIS SHOULD NOT FAIL AS THERE SHOULD BE ONLY 1 TRANSACTION!
+ assertEquals(1, getQueueSize(holdKahaDb.getQueueName()));
+ }
+
+ @Test
+ public void testAMQ7067rollback() throws Exception {
+ final Connection connection = ACTIVE_MQ_NON_XA_CONNECTION_FACTORY.createConnection();
+ connection.start();
+
+ Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
+ Queue holdKahaDb = session.createQueue("holdKahaDb");
+ MessageProducer holdKahaDbProducer = session.createProducer(holdKahaDb);
+ TextMessage helloMessage = session.createTextMessage(StringUtils.repeat("a", 10));
+ holdKahaDbProducer.send(helloMessage);
+ Queue queue = session.createQueue("test");
+ produce(connection, queue, 100, 512*1024);
+ session.rollback();
+ produce(connection, queue, 100, 512*1024);
+
+ System.out.println(String.format("QueueSize %s: %d", holdKahaDb.getQueueName(), getQueueSize(holdKahaDb.getQueueName())));
+ purgeQueue(queue.getQueueName());
+ Thread.sleep(10000);
+
+ curruptIndexFile(getDataDirectory());
+
+
+ while(true) {
+ try {
+ Thread.sleep(10000);
+ System.out.println(String.format("QueueSize %s: %d", holdKahaDb.getQueueName(), getQueueSize(holdKahaDb.getQueueName())));
+ break;
+ } catch (Exception ex) {
+ System.out.println(ex.getMessage());
+ break;
+ }
+ }
+
+ connection.close();
+
+ // THIS SHOULD NOT FAIL AS THERE SHOULD ZERO TRANSACTION!
+ assertEquals(0, getQueueSize(holdKahaDb.getQueueName()));
+ }
+
+ protected static void createDanglingTransaction(XAResource xaRes, XASession xaSession, Queue queue) throws JMSException, IOException, XAException {
+ MessageProducer producer = xaSession.createProducer(queue);
+ XATransactionId txId = createXATransaction();
+ xaRes.start(txId, TMNOFLAGS);
+
+ TextMessage helloMessage = xaSession.createTextMessage(StringUtils.repeat("dangler", 10));
+ producer.send(helloMessage);
+ xaRes.end(txId, TMSUCCESS);
+ xaRes.prepare(txId);
+ System.out.println("****** createDanglingTransaction txId = " + txId);
+ }
+
+ protected static void produce(XAResource xaRes, XASession xaSession, Queue queue, int messageCount, int messageSize) throws JMSException, IOException, XAException {
+ MessageProducer producer = xaSession.createProducer(queue);
+
+ for (int i = 0; i < messageCount; i++) {
+ XATransactionId txid = createXATransaction();
+ xaRes.start(txid, TMNOFLAGS);
+
+ TextMessage helloMessage = xaSession.createTextMessage(StringUtils.repeat("a", messageSize));
+ producer.send(helloMessage);
+ xaRes.end(txid, TMSUCCESS);
+ xaRes.prepare(txid);
+ xaRes.commit(txid, true);
+ }
+ }
+
+ protected static void produce(Connection connection, Queue queue, int messageCount, int messageSize) throws JMSException, IOException, XAException {
+ Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(queue);
+
+ for (int i = 0; i < messageCount; i++) {
+ TextMessage helloMessage = session.createTextMessage(StringUtils.repeat("a", messageSize));
+ producer.send(helloMessage);
+ session.commit();
+
+ }
+ }
+
+ protected static XATransactionId createXATransaction() throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream os = new DataOutputStream(baos);
+ os.writeLong(r.nextInt());
+ os.close();
+ byte[] bs = baos.toByteArray();
+
+ XATransactionId xid = new XATransactionId();
+ xid.setBranchQualifier(bs);
+ xid.setGlobalTransactionId(bs);
+ xid.setFormatId(55);
+ return xid;
+ }
+
+ private RecoveredXATransactionViewMBean getProxyToPreparedTransactionViewMBean(TransactionId xid) throws MalformedObjectNameException, JMSException {
+
+ ObjectName objectName = new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost,transactionType=RecoveredXaTransaction,xid=" +
+ JMXSupport.encodeObjectNamePart(xid.toString()));
+ RecoveredXATransactionViewMBean proxy = (RecoveredXATransactionViewMBean) broker.getManagementContext().newProxyInstance(objectName,
+ RecoveredXATransactionViewMBean.class, true);
+ return proxy;
+ }
+
+ private PersistenceAdapterViewMBean getProxyToPersistenceAdapter(String name) throws MalformedObjectNameException, JMSException {
+ return (PersistenceAdapterViewMBean) broker.getManagementContext().newProxyInstance(
+ BrokerMBeanSupport.createPersistenceAdapterName(broker.getBrokerObjectName().toString(), name),
+ PersistenceAdapterViewMBean.class, true);
+ }
+
+ private void deleteData(File file) throws Exception {
+ String[] entries = file.list();
+ if (entries == null) return;
+ for (String s : entries) {
+ File currentFile = new File(file.getPath(), s);
+ if (currentFile.isDirectory()) {
+ deleteData(currentFile);
+ }
+ currentFile.delete();
+ }
+ file.delete();
+ }
+
+ private long getQueueSize(final String queueName) throws MalformedObjectNameException {
+ ObjectName objectName = new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost,destinationType=Queue,destinationName=" + JMXSupport.encodeObjectNamePart(queueName));
+ DestinationViewMBean proxy = (DestinationViewMBean) broker.getManagementContext().newProxyInstance(objectName, DestinationViewMBean.class, true);
+ return proxy.getQueueSize();
+ }
+
+ private void purgeQueue(final String queueName) throws MalformedObjectNameException, Exception {
+ ObjectName objectName = new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost,destinationType=Queue,destinationName=" + JMXSupport.encodeObjectNamePart(queueName));
+ QueueViewMBean proxy = (QueueViewMBean) broker.getManagementContext().newProxyInstance(objectName, QueueViewMBean.class, true);
+ proxy.purge();
+ }
+
+ private String getDataDirectory() throws MalformedObjectNameException {
+ ObjectName objectName = new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost");
+ BrokerViewMBean proxy = (BrokerViewMBean) broker.getManagementContext().newProxyInstance(objectName, BrokerViewMBean.class, true);
+ return proxy.getDataDirectory();
+ }
+
+ protected static void curruptIndexFile(final String dataPath) throws FileNotFoundException, UnsupportedEncodingException {
+ PrintWriter writer = new PrintWriter(String.format("%s/kahadb/db.data", dataPath), "UTF-8");
+ writer.println("asdasdasd");
+ writer.close();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/activemq/blob/99db9ef3/activemq-unit-tests/src/test/resources/org/apache/activemq/bugs/amq7067/activemq.xml
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/resources/org/apache/activemq/bugs/amq7067/activemq.xml b/activemq-unit-tests/src/test/resources/org/apache/activemq/bugs/amq7067/activemq.xml
new file mode 100644
index 0000000..70da26b
--- /dev/null
+++ b/activemq-unit-tests/src/test/resources/org/apache/activemq/bugs/amq7067/activemq.xml
@@ -0,0 +1,111 @@
+<!--
+ 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.
+-->
+<!-- START SNIPPET: example -->
+<beans
+ xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+ http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
+
+ <!--
+ The <broker> element is used to configure the ActiveMQ broker.
+ -->
+ <broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="target/data">
+
+ <destinationPolicy>
+ <policyMap>
+ <policyEntries>
+ <policyEntry topic=">" >
+ <!-- The constantPendingMessageLimitStrategy is used to prevent
+ slow topic consumers to block producers and affect other consumers
+ by limiting the number of messages that are retained
+ For more information, see:
+
+ http://activemq.apache.org/slow-consumer-handling.html
+
+ -->
+ <pendingMessageLimitStrategy>
+ <constantPendingMessageLimitStrategy limit="1000"/>
+ </pendingMessageLimitStrategy>
+ </policyEntry>
+ </policyEntries>
+ </policyMap>
+ </destinationPolicy>
+
+
+ <!--
+ The managementContext is used to configure how ActiveMQ is exposed in
+ JMX. By default, ActiveMQ uses the MBean server that is started by
+ the JVM. For more information, see:
+
+ http://activemq.apache.org/jmx.html
+ -->
+ <managementContext>
+ <managementContext createConnector="false"/>
+ </managementContext>
+
+ <!--
+ Configure message persistence for the broker. The default persistence
+ mechanism is the KahaDB store (identified by the kahaDB tag).
+ For more information, see:
+
+ http://activemq.apache.org/persistence.html
+ -->
+ <persistenceAdapter>
+ <kahaDB directory="target/data/kahadb" />
+ </persistenceAdapter>
+
+
+ <!--
+ The systemUsage controls the maximum amount of space the broker will
+ use before disabling caching and/or slowing down producers. For more information, see:
+ http://activemq.apache.org/producer-flow-control.html
+ -->
+ <systemUsage>
+ <systemUsage>
+ <memoryUsage>
+ <memoryUsage percentOfJvmHeap="70" />
+ </memoryUsage>
+ <storeUsage>
+ <storeUsage limit="100 gb"/>
+ </storeUsage>
+ <tempUsage>
+ <tempUsage limit="50 gb"/>
+ </tempUsage>
+ </systemUsage>
+ </systemUsage>
+
+ <!--
+ The transport connectors expose ActiveMQ over a given protocol to
+ clients and other brokers. For more information, see:
+
+ http://activemq.apache.org/configuring-transports.html
+ -->
+ <transportConnectors>
+ <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
+ <transportConnector name="openwire" uri="tcp://localhost:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
+ </transportConnectors>
+
+ <!-- destroy the spring context on shutdown to stop jetty -->
+ <shutdownHooks>
+ <bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" />
+ </shutdownHooks>
+
+ </broker>
+
+</beans>
+ <!-- END SNIPPET: example -->
\ No newline at end of file
[2/2] activemq git commit: AMQ-7067 - tidy up tests and add prepare
variant,
limit rollback location recording to xa case. There is still some work to do
for the ack compaction case to make it aware of the tx records such that
those are transferred as ne
Posted by gt...@apache.org.
AMQ-7067 - tidy up tests and add prepare variant, limit rollback location recording to xa case. There is still some work to do for the ack compaction case to make it aware of the tx records such that those are transferred as necessary
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/57c79395
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/57c79395
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/57c79395
Branch: refs/heads/master
Commit: 57c7939534a927bfc2d1b0454aac7ef8d804532b
Parents: 99db9ef
Author: gtully <ga...@gmail.com>
Authored: Tue Oct 9 12:01:47 2018 +0100
Committer: gtully <ga...@gmail.com>
Committed: Tue Oct 9 12:01:47 2018 +0100
----------------------------------------------------------------------
.../activemq/store/kahadb/MessageDatabase.java | 2 +-
.../org/apache/activemq/bugs/AMQ7067Test.java | 128 ++++++++++++++-----
2 files changed, 99 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/57c79395/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 1c974cc..46696f4 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
@@ -1437,7 +1437,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
updates = preparedTransactions.remove(key);
}
}
- if (updates != null) {
+ if (key.isXATransaction() && updates != null) {
for(Operation op : updates) {
recordAckMessageReferenceLocation(location, op.getLocation());
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/57c79395/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ7067Test.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ7067Test.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ7067Test.java
index c1f34d0..d00ee41 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ7067Test.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ7067Test.java
@@ -14,12 +14,14 @@ import org.apache.activemq.broker.jmx.RecoveredXATransactionViewMBean;
import org.apache.activemq.command.TransactionId;
import org.apache.activemq.command.XATransactionId;
import org.apache.activemq.util.JMXSupport;
+import org.apache.activemq.util.Wait;
import org.apache.commons.lang.StringUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import javax.jms.*;
+import javax.management.InstanceNotFoundException;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import javax.transaction.xa.XAException;
@@ -32,11 +34,14 @@ import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.UnsupportedEncodingException;
-import java.net.URI;
+import java.lang.reflect.UndeclaredThrowableException;
import java.util.Random;
+import java.util.concurrent.TimeUnit;
import static javax.transaction.xa.XAResource.*;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
public class AMQ7067Test {
@@ -82,11 +87,55 @@ public class AMQ7067Test {
broker.start();
}
+ @Test
+ public void testXAPrepare() throws Exception {
+
+ setupXAConnection();
+
+ Queue holdKahaDb = xaSession.createQueue("holdKahaDb");
+
+ MessageProducer holdKahaDbProducer = xaSession.createProducer(holdKahaDb);
+
+ XATransactionId txid = createXATransaction();
+ System.out.println("****** create new txid = " + txid);
+ xaRes.start(txid, TMNOFLAGS);
+
+ TextMessage helloMessage = xaSession.createTextMessage(StringUtils.repeat("a", 10));
+ holdKahaDbProducer.send(helloMessage);
+ xaRes.end(txid, TMSUCCESS);
+
+ Queue queue = xaSession.createQueue("test");
+
+ produce(xaRes, xaSession, queue, 100, 512 * 1024);
+
+ xaRes.prepare(txid);
+
+ produce(xaRes, xaSession, queue, 100, 512 * 1024);
+
+ ((org.apache.activemq.broker.region.Queue) broker.getRegionBroker().getDestinationMap().get(queue)).purge();
+
+ Xid[] xids = xaRes.recover(TMSTARTRSCAN);
+
+ //Should be 1 since we have only 1 prepared
+ assertEquals(1, xids.length);
+ connection.close();
+
+ broker.stop();
+ broker.waitUntilStopped();
+ createBroker();
+
+ setupXAConnection();
+ xids = xaRes.recover(TMSTARTRSCAN);
+
+ System.out.println("****** recovered = " + xids);
+
+ // THIS SHOULD NOT FAIL AS THERE SHOULD DBE ONLY 1 TRANSACTION!
+ assertEquals(1, xids.length);
+ }
@Test
- public void testAMQ7067XAcommit() throws Exception {
+ public void testXAcommit() throws Exception {
- PersistenceAdapterViewMBean kahadbView = getProxyToPersistenceAdapter(broker.getPersistenceAdapter().toString());
setupXAConnection();
Queue holdKahaDb = xaSession.createQueue("holdKahaDb");
@@ -124,15 +173,14 @@ public class AMQ7067Test {
setupXAConnection();
xids = xaRes.recover(TMSTARTRSCAN);
- // THIS SHOULD NOT FAIL AS THERE SHOUL DBE ONLY 1 TRANSACTION!
+ // THIS SHOULD NOT FAIL AS THERE SHOULD DBE ONLY 1 TRANSACTION!
assertEquals(1, xids.length);
}
@Test
- public void testAMQ7067XArollback() throws Exception {
+ public void testXArollback() throws Exception {
- PersistenceAdapterViewMBean kahadbView = getProxyToPersistenceAdapter(broker.getPersistenceAdapter().toString());
setupXAConnection();
Queue holdKahaDb = xaSession.createQueue("holdKahaDb");
@@ -176,11 +224,11 @@ public class AMQ7067Test {
}
@Test
- public void testAMQ7067commit() throws Exception {
+ public void testCommit() throws Exception {
final Connection connection = ACTIVE_MQ_NON_XA_CONNECTION_FACTORY.createConnection();
connection.start();
- Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
+ Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
Queue holdKahaDb = session.createQueue("holdKahaDb");
MessageProducer holdKahaDbProducer = session.createProducer(holdKahaDb);
TextMessage helloMessage = session.createTextMessage(StringUtils.repeat("a", 10));
@@ -192,14 +240,28 @@ public class AMQ7067Test {
System.out.println(String.format("QueueSize %s: %d", holdKahaDb.getQueueName(), getQueueSize(holdKahaDb.getQueueName())));
purgeQueue(queue.getQueueName());
- Thread.sleep(10000);
+ Wait.waitFor(new Wait.Condition() {
+ @Override
+ public boolean isSatisified() throws Exception {
+ return 0 == getQueueSize(queue.getQueueName());
+ }
+ });
+
+ // force gc
+ broker.getPersistenceAdapter().checkpoint(true);
+
+ connection.close();
curruptIndexFile(getDataDirectory());
+ broker.stop();
+ broker.waitUntilStopped();
+ createBroker();
+ broker.waitUntilStarted();
while(true) {
try {
- Thread.sleep(10000);
+ TimeUnit.SECONDS.sleep(1);
System.out.println(String.format("QueueSize %s: %d", holdKahaDb.getQueueName(), getQueueSize(holdKahaDb.getQueueName())));
break;
} catch (Exception ex) {
@@ -208,18 +270,16 @@ public class AMQ7067Test {
}
}
- connection.close();
-
// THIS SHOULD NOT FAIL AS THERE SHOULD BE ONLY 1 TRANSACTION!
assertEquals(1, getQueueSize(holdKahaDb.getQueueName()));
}
@Test
- public void testAMQ7067rollback() throws Exception {
+ public void testRollback() throws Exception {
final Connection connection = ACTIVE_MQ_NON_XA_CONNECTION_FACTORY.createConnection();
connection.start();
- Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
+ Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
Queue holdKahaDb = session.createQueue("holdKahaDb");
MessageProducer holdKahaDbProducer = session.createProducer(holdKahaDb);
TextMessage helloMessage = session.createTextMessage(StringUtils.repeat("a", 10));
@@ -231,26 +291,34 @@ public class AMQ7067Test {
System.out.println(String.format("QueueSize %s: %d", holdKahaDb.getQueueName(), getQueueSize(holdKahaDb.getQueueName())));
purgeQueue(queue.getQueueName());
- Thread.sleep(10000);
-
- curruptIndexFile(getDataDirectory());
-
- while(true) {
- try {
- Thread.sleep(10000);
- System.out.println(String.format("QueueSize %s: %d", holdKahaDb.getQueueName(), getQueueSize(holdKahaDb.getQueueName())));
- break;
- } catch (Exception ex) {
- System.out.println(ex.getMessage());
- break;
+ Wait.waitFor(new Wait.Condition() {
+ @Override
+ public boolean isSatisified() throws Exception {
+ return 0 == getQueueSize(queue.getQueueName());
}
- }
+ });
+
+ // force gc
+ broker.getPersistenceAdapter().checkpoint(true);
connection.close();
+ curruptIndexFile(getDataDirectory());
+
+ broker.stop();
+ broker.waitUntilStopped();
+ createBroker();
+ broker.waitUntilStarted();
- // THIS SHOULD NOT FAIL AS THERE SHOULD ZERO TRANSACTION!
- assertEquals(0, getQueueSize(holdKahaDb.getQueueName()));
+
+ // no sign of the test queue on recovery, rollback is the default for any inflight
+ // this test serves as a sanity check on existing behaviour
+ try {
+ getQueueSize(holdKahaDb.getQueueName());
+ fail("expect InstanceNotFoundException");
+ } catch (UndeclaredThrowableException expected) {
+ assertTrue(expected.getCause() instanceof InstanceNotFoundException);
+ }
}
protected static void createDanglingTransaction(XAResource xaRes, XASession xaSession, Queue queue) throws JMSException, IOException, XAException {
@@ -281,7 +349,7 @@ public class AMQ7067Test {
}
protected static void produce(Connection connection, Queue queue, int messageCount, int messageSize) throws JMSException, IOException, XAException {
- Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
+ Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
MessageProducer producer = session.createProducer(queue);
for (int i = 0; i < messageCount; i++) {