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/05/18 09:32:51 UTC
activemq git commit: AMQ-6707 - ensure entryLocator is used for
rollback of prepared add to avoid NPE, relates to AMQ-5567
Repository: activemq
Updated Branches:
refs/heads/master 1db36d864 -> 72613aaba
AMQ-6707 - ensure entryLocator is used for rollback of prepared add to avoid NPE, relates to AMQ-5567
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/72613aab
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/72613aab
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/72613aab
Branch: refs/heads/master
Commit: 72613aaba6d2641b8635e54e2fb4a28c14da2021
Parents: 1db36d8
Author: gtully <ga...@gmail.com>
Authored: Fri May 18 10:32:00 2018 +0100
Committer: gtully <ga...@gmail.com>
Committed: Fri May 18 10:32:00 2018 +0100
----------------------------------------------------------------------
.../store/jdbc/JDBCPersistenceAdapter.java | 2 +-
.../activemq/store/jdbc/XACompletionTest.java | 130 +++++++++++++++++++
2 files changed, 131 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/72613aab/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCPersistenceAdapter.java
----------------------------------------------------------------------
diff --git a/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCPersistenceAdapter.java b/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCPersistenceAdapter.java
index 95b1446..c193c46 100644
--- a/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCPersistenceAdapter.java
+++ b/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCPersistenceAdapter.java
@@ -800,7 +800,7 @@ public class JDBCPersistenceAdapter extends DataSourceServiceSupport implements
public void commitRemove(ConnectionContext context, MessageAck ack) throws IOException {
TransactionContext c = getTransactionContext(context);
try {
- getAdapter().doRemoveMessage(c, (Long)ack.getLastMessageId().getFutureOrSequenceLong(), null);
+ getAdapter().doRemoveMessage(c, (Long)ack.getLastMessageId().getEntryLocator(), null);
} catch (SQLException e) {
JDBCPersistenceAdapter.log("JDBC Failure: ", e);
throw IOExceptionSupport.create("Failed to commit last ack: " + ack + ". Reason: " + e,e);
http://git-wip-us.apache.org/repos/asf/activemq/blob/72613aab/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/XACompletionTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/XACompletionTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/XACompletionTest.java
index ee3a88c..50cb1c9 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/XACompletionTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/XACompletionTest.java
@@ -619,6 +619,136 @@ public class XACompletionTest extends TestSupport {
}
+ @Test
+ public void testMoveInTwoBranchesPreparedAckRecoveryRestartRollback() throws Exception {
+
+ factory = new ActiveMQXAConnectionFactory(
+ connectionUri + "?jms.prefetchPolicy.all=0&jms.redeliveryPolicy.maximumRedeliveries=" + messagesExpected);
+
+ factory.setWatchTopicAdvisories(false);
+ sendMessages(messagesExpected);
+
+ ActiveMQXAConnection activeMQXAConnection = (ActiveMQXAConnection) factory.createXAConnection();
+ activeMQXAConnection.start();
+ XASession xaSession = activeMQXAConnection.createXASession();
+
+ Destination destination = xaSession.createQueue("TEST");
+ MessageConsumer consumer = xaSession.createConsumer(destination);
+
+ XAResource resource = xaSession.getXAResource();
+
+ final Xid tid = createXid();
+ byte[] branch = tid.getBranchQualifier();
+ final byte[] branch2 = Arrays.copyOf(branch, branch.length);
+ branch2[0] = '!';
+
+ Xid branchTid = new Xid() {
+ @Override
+ public int getFormatId() {
+ return tid.getFormatId();
+ }
+
+ @Override
+ public byte[] getGlobalTransactionId() {
+ return tid.getGlobalTransactionId();
+ }
+
+ @Override
+ public byte[] getBranchQualifier() {
+ return branch2;
+ }
+ };
+
+ resource.start(tid, XAResource.TMNOFLAGS);
+
+ int messagesReceived = 0;
+
+ Message message = null;
+
+ for (int i = 0; i < messagesExpected; i++) {
+
+ try {
+ LOG.debug("Receiving message " + (messagesReceived + 1) + " of " + messagesExpected);
+ message = consumer.receive(2000);
+ LOG.info("Received : " + message);
+ messagesReceived++;
+ } catch (Exception e) {
+ LOG.debug("Caught exception:", e);
+ }
+ }
+
+ resource.end(tid, XAResource.TMSUCCESS);
+
+ ActiveMQXAConnection activeMQXAConnectionSend = (ActiveMQXAConnection) factory.createXAConnection();
+ activeMQXAConnectionSend.start();
+ XASession xaSessionSend = activeMQXAConnection.createXASession();
+
+ Destination destinationSend = xaSessionSend.createQueue("TEST_MOVE");
+ MessageProducer producer = xaSessionSend.createProducer(destinationSend);
+
+ XAResource resourceSend = xaSessionSend.getXAResource();
+ resourceSend.start(branchTid, XAResource.TMNOFLAGS);
+
+ ActiveMQMessage toSend = (ActiveMQMessage) xaSessionSend.createTextMessage();
+ toSend.setTransactionId(new XATransactionId(branchTid));
+ producer.send(toSend);
+
+ resourceSend.end(branchTid, XAResource.TMSUCCESS);
+ resourceSend.prepare(branchTid);
+
+ // ack on TEST is prepared
+ resource.prepare(tid);
+
+ // send to TEST_MOVE is rolledback
+ resourceSend.rollback(branchTid);
+
+ consumer.close();
+
+ LOG.info("Prepared");
+ dumpMessages();
+
+ broker = restartBroker();
+
+ LOG.info("New broker");
+ dumpMessages();
+
+ ObjectName queueViewMBeanName = new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost,destinationType=Queue,destinationName=TEST");
+ QueueViewMBean proxy = (QueueViewMBean) broker.getManagementContext()
+ .newProxyInstance(queueViewMBeanName, QueueViewMBean.class, true);
+
+ assertEquals("size", 0, proxy.getQueueSize());
+
+ assertNull(regularReceive("TEST_MOVE"));
+
+ ObjectName queueMoveViewMBeanName = new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost,destinationType=Queue,destinationName=TEST_MOVE");
+ QueueViewMBean moveProxy = (QueueViewMBean) broker.getManagementContext()
+ .newProxyInstance(queueMoveViewMBeanName, QueueViewMBean.class, true);
+
+ assertEquals("enq", 0, moveProxy.getDequeueCount());
+ assertEquals("size", 0, moveProxy.getQueueSize());
+
+ assertEquals("size 0", 0, moveProxy.getQueueSize());
+
+ factory = new ActiveMQXAConnectionFactory(
+ connectionUri + "?jms.prefetchPolicy.all=0&jms.redeliveryPolicy.maximumRedeliveries=" + messagesExpected);
+ factory.setWatchTopicAdvisories(false);
+
+ activeMQXAConnection = (ActiveMQXAConnection) factory.createXAConnection();
+ activeMQXAConnection.start();
+ xaSession = activeMQXAConnection.createXASession();
+
+ resource = xaSession.getXAResource();
+ resource.rollback(tid);
+
+ assertEquals("size", 1, proxy.getQueueSize());
+ assertEquals("c size", 1, proxy.cursorSize());
+
+ assertNotNull(regularReceive("TEST"));
+
+ assertEquals("size", 0, proxy.getQueueSize());
+ assertEquals("c size", 0, proxy.cursorSize());
+ assertEquals("dq", 1, proxy.getDequeueCount());
+ }
@Test