Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

https://issues.apache.org/jira/browse/AMQ-6089 - support TMNOFLAGS as a scan end to allow looping calls to recover to terminate

This commit is contained in:
gtully 2016-01-06 12:54:20 +00:00
parent e3df09b9db
commit 16bc0f0d75
2 changed files with 38 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
activemq-client/src/main/java/org/apache/activemq/TransactionContext.java
View File

@ -672,27 +672,33 @@ public class TransactionContext implements XAResource {
@Override @Override
public Xid[] recover(int flag) throws XAException { public Xid[] recover(int flag) throws XAException {
LOG.debug("recover({})", flag); LOG.debug("recover({})", flag);
XATransactionId[] answer;
if (XAResource.TMNOFLAGS == flag) {
// signal next in cursor scan, which for us is always the end b/c we don't maintain any cursor state
// allows looping scan to complete
answer = new XATransactionId[0];
} else {
TransactionInfo info = new TransactionInfo(getConnectionId(), null, TransactionInfo.RECOVER); TransactionInfo info = new TransactionInfo(getConnectionId(), null, TransactionInfo.RECOVER);
try { try {
this.connection.checkClosedOrFailed(); this.connection.checkClosedOrFailed();
this.connection.ensureConnectionInfoSent(); this.connection.ensureConnectionInfoSent();
DataArrayResponse receipt = (DataArrayResponse)this.connection.syncSendPacket(info); DataArrayResponse receipt = (DataArrayResponse) this.connection.syncSendPacket(info);
DataStructure[] data = receipt.getData(); DataStructure[] data = receipt.getData();
XATransactionId[] answer;
if (data instanceof XATransactionId[]) { if (data instanceof XATransactionId[]) {
answer = (XATransactionId[])data; answer = (XATransactionId[]) data;
} else { } else {
answer = new XATransactionId[data.length]; answer = new XATransactionId[data.length];
System.arraycopy(data, 0, answer, 0, data.length); System.arraycopy(data, 0, answer, 0, data.length);
} }
LOG.debug("recover({})={}", flag, answer);
return answer;
} catch (JMSException e) { } catch (JMSException e) {
throw toXAException(e); throw toXAException(e);
} }
} }
LOG.debug("recover({})={}", flag, answer);
return answer;
}
@Override @Override
public int getTransactionTimeout() throws XAException { public int getTransactionTimeout() throws XAException {

16
activemq-unit-tests/src/test/java/org/apache/activemq/broker/XARecoveryBrokerTest.java
View File

@ -16,7 +16,9 @@
*/ */
package org.apache.activemq.broker; package org.apache.activemq.broker;
import java.util.Arrays;
import java.util.HashSet; import java.util.HashSet;
import java.util.LinkedList;
import java.util.Set; import java.util.Set;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
@ -24,7 +26,11 @@ import javax.jms.JMSException;
import javax.management.InstanceNotFoundException; import javax.management.InstanceNotFoundException;
import javax.management.MalformedObjectNameException; import javax.management.MalformedObjectNameException;
import javax.management.ObjectName; import javax.management.ObjectName;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;
import junit.framework.Test; import junit.framework.Test;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.TransactionContext;
import org.apache.activemq.broker.jmx.BrokerMBeanSupport; import org.apache.activemq.broker.jmx.BrokerMBeanSupport;
import org.apache.activemq.broker.jmx.DestinationViewMBean; import org.apache.activemq.broker.jmx.DestinationViewMBean;
import org.apache.activemq.broker.jmx.PersistenceAdapterViewMBean; import org.apache.activemq.broker.jmx.PersistenceAdapterViewMBean;
@ -102,6 +108,16 @@ public class XARecoveryBrokerTest extends BrokerRestartTestSupport {
dar = (DataArrayResponse)response; dar = (DataArrayResponse)response;
assertEquals(4, dar.getData().length); assertEquals(4, dar.getData().length);
// verify XAResource scan loop
XAResource transactionContextXAResource = new TransactionContext(ActiveMQConnection.makeConnection(broker.getVmConnectorURI().toString()));
LinkedList<Xid> tracked = new LinkedList<Xid>();
Xid[] recoveryXids = transactionContextXAResource.recover(XAResource.TMSTARTRSCAN);
while (recoveryXids.length > 0) {
tracked.addAll(Arrays.asList(recoveryXids));
recoveryXids = transactionContextXAResource.recover(XAResource.TMNOFLAGS);
}
assertEquals("got 4 via scan loop", 4, tracked.size());
// validate destination depth via jmx // validate destination depth via jmx
DestinationViewMBean destinationView = getProxyToDestination(destinationList(destination)[0]); DestinationViewMBean destinationView = getProxyToDestination(destinationList(destination)[0]);
assertEquals("enqueue count does not see prepared", 0, destinationView.getQueueSize()); assertEquals("enqueue count does not see prepared", 0, destinationView.getQueueSize());