diff options
6 files changed, 72 insertions, 40 deletions
diff --git a/java/010ExcludeList b/java/010ExcludeList index ca0ad31412..59a4a0097c 100644 --- a/java/010ExcludeList +++ b/java/010ExcludeList @@ -45,5 +45,5 @@ org.apache.qpid.test.testcases.MandatoryMessageTest#test_QPID_508_MandatoryFails org.apache.qpid.test.testcases.MandatoryMessageTest#test_QPID_508_MandatoryFailsNoRouteTxP2P org.apache.qpid.test.testcases.MandatoryMessageTest#test_QPID_508_MandatoryFailsNoRouteNoTxPubSub org.apache.qpid.test.testcases.MandatoryMessageTest#test_QPID_508_MandatoryFailsNoRouteTxPubSub -// Those tests are failing, they must be removed from this list once QPID-1095 is fixed. -org.apache.qpid.test.unit.xa.FaultTest#*
\ No newline at end of file +// the 0.10 c++ broker does not implement forget +org.apache.qpid.test.unit.xa.FaultTest#testForget
\ No newline at end of file diff --git a/java/010ExcludeList-store b/java/010ExcludeList-store index 1692fd2345..ccd47ce58b 100644 --- a/java/010ExcludeList-store +++ b/java/010ExcludeList-store @@ -40,5 +40,5 @@ org.apache.qpid.test.testcases.MandatoryMessageTest#test_QPID_508_MandatoryFails org.apache.qpid.test.testcases.MandatoryMessageTest#test_QPID_508_MandatoryFailsNoRouteTxP2P org.apache.qpid.test.testcases.MandatoryMessageTest#test_QPID_508_MandatoryFailsNoRouteNoTxPubSub org.apache.qpid.test.testcases.MandatoryMessageTest#test_QPID_508_MandatoryFailsNoRouteTxPubSub -// Those tests are failing, they must be removed from this list once QPID-1095 is fixed. -org.apache.qpid.test.unit.xa.FaultTest#*
\ No newline at end of file +// the 0.10 c++ broker does not implement forget +org.apache.qpid.test.unit.xa.FaultTest#testForget
\ No newline at end of file diff --git a/java/client/src/main/java/org/apache/qpid/client/XAResourceImpl.java b/java/client/src/main/java/org/apache/qpid/client/XAResourceImpl.java index 27ec445436..7a107d748b 100644 --- a/java/client/src/main/java/org/apache/qpid/client/XAResourceImpl.java +++ b/java/client/src/main/java/org/apache/qpid/client/XAResourceImpl.java @@ -116,6 +116,17 @@ public class XAResourceImpl implements XAResource { _logger.debug("end tx branch with xid: ", xid); } + switch (flag) + { + case(XAResource.TMSUCCESS): + break; + case(XAResource.TMFAIL): + break; + case(XAResource.TMSUSPEND): + break; + default: + throw new XAException(XAException.XAER_INVAL); + } Future<XaResult> future = _xaSession.getQpidSession() .dtxEnd(convertXid(xid), flag == XAResource.TMFAIL ? Option.FAIL : Option.NO_OPTION, @@ -270,8 +281,18 @@ public class XAResourceImpl implements XAResource { // the flag is ignored Future<RecoverResult> future = _xaSession.getQpidSession().dtxRecover(); - RecoverResult res = future.get(); - // todo make sure that the keys of the returned map are the xids + RecoverResult res = null; + try + { + res = future.get(); + } + catch (SessionException e) + { + // we need to restore the qpidity session that has been closed + _xaSession.createSession(); + // we should get a single exception + convertExecutionErrorToXAErr( e.getExceptions().get(0).getErrorCode()); + } Xid[] result = new Xid[res.getInDoubt().size()]; int i = 0; for (Object obj : res.getInDoubt()) @@ -366,6 +387,17 @@ public class XAResourceImpl implements XAResource { _logger.debug("start tx branch with xid: ", xid); } + switch (flag) + { + case(XAResource.TMNOFLAGS): + break; + case(XAResource.TMJOIN): + break; + case(XAResource.TMRESUME): + break; + default: + throw new XAException(XAException.XAER_INVAL); + } Future<XaResult> future = _xaSession.getQpidSession() .dtxStart(convertXid(xid), flag == XAResource.TMJOIN ? Option.JOIN : Option.NO_OPTION, diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/xa/FaultTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/xa/FaultTest.java index 2c99ce707d..49a434518d 100644 --- a/java/client/src/test/java/org/apache/qpid/test/unit/xa/FaultTest.java +++ b/java/client/src/test/java/org/apache/qpid/test/unit/xa/FaultTest.java @@ -359,7 +359,7 @@ public class FaultTest extends AbstractXATestCase try { _xaResource.start(xid, XAResource.TMNOFLAGS); - _xaResource.end(xid, XAResource.TMNOFLAGS); + _xaResource.end(xid, XAResource.TMSUCCESS); _xaResource.commit(xid, false); fail("We managed to commit a not prepared xid"); } @@ -375,7 +375,7 @@ public class FaultTest extends AbstractXATestCase try { _xaResource.start(xid, XAResource.TMNOFLAGS); - _xaResource.end(xid, XAResource.TMNOFLAGS); + _xaResource.end(xid, XAResource.TMSUCCESS); _xaResource.prepare(xid); _xaResource.commit(xid, true); fail("We managed to commit a prepared xid"); @@ -444,7 +444,7 @@ public class FaultTest extends AbstractXATestCase assertEquals("Wrong timeout", _xaResource.getTransactionTimeout(), 0); _xaResource.setTransactionTimeout(1000); assertEquals("Wrong timeout", _xaResource.getTransactionTimeout(), 1000); - _xaResource.end(xid, XAResource.TMNOFLAGS); + _xaResource.end(xid, XAResource.TMSUCCESS); xid = getNewXid(); _xaResource.start(xid, XAResource.TMNOFLAGS); assertEquals("Wrong timeout", _xaResource.getTransactionTimeout(), 0); @@ -472,7 +472,7 @@ public class FaultTest extends AbstractXATestCase assertEquals("Wrong timeout", _xaResource.getTransactionTimeout(), 0); _xaResource.setTransactionTimeout(10); Thread.sleep(1000); - _xaResource.end(xid, XAResource.TMNOFLAGS); + _xaResource.end(xid, XAResource.TMSUCCESS); } catch (XAException e) { diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/xa/QueueTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/xa/QueueTest.java index e786981183..740f9e72ad 100644 --- a/java/client/src/test/java/org/apache/qpid/test/unit/xa/QueueTest.java +++ b/java/client/src/test/java/org/apache/qpid/test/unit/xa/QueueTest.java @@ -181,7 +181,7 @@ public class QueueTest extends AbstractXATestCase // start the xaResource for xid1 try { - _xaResource.start(xid1, XAResource.TMSUCCESS); + _xaResource.start(xid1, XAResource.TMNOFLAGS); } catch (XAException e) { @@ -212,7 +212,7 @@ public class QueueTest extends AbstractXATestCase // start the xaResource for xid2 try { - _xaResource.start(xid2, XAResource.TMSUCCESS); + _xaResource.start(xid2, XAResource.TMNOFLAGS); } catch (XAException e) { @@ -331,7 +331,7 @@ public class QueueTest extends AbstractXATestCase // start the xaResource for xid1 try { - _xaResource.start(xid1, XAResource.TMSUCCESS); + _xaResource.start(xid1, XAResource.TMNOFLAGS); } catch (XAException e) { @@ -462,7 +462,7 @@ public class QueueTest extends AbstractXATestCase // start the xaResource for xid1 try { - _xaResource.start(xid1, XAResource.TMSUCCESS); + _xaResource.start(xid1, XAResource.TMNOFLAGS); } catch (XAException e) { @@ -519,7 +519,7 @@ public class QueueTest extends AbstractXATestCase // start the xaResource for xid2 try { - _xaResource.start(xid2, XAResource.TMSUCCESS); + _xaResource.start(xid2, XAResource.TMNOFLAGS); } catch (XAException e) { diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/xa/TopicTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/xa/TopicTest.java index 04fd8b8c8d..027257d761 100644 --- a/java/client/src/test/java/org/apache/qpid/test/unit/xa/TopicTest.java +++ b/java/client/src/test/java/org/apache/qpid/test/unit/xa/TopicTest.java @@ -193,7 +193,7 @@ public class TopicTest extends AbstractXATestCase try { _logger.debug("starting tx branch xid1"); - _xaResource.start(xid1, XAResource.TMSUCCESS); + _xaResource.start(xid1, XAResource.TMNOFLAGS); } catch (XAException e) { @@ -224,7 +224,7 @@ public class TopicTest extends AbstractXATestCase _logger.debug("start the xaResource for xid2"); try { - _xaResource.start(xid2, XAResource.TMSUCCESS); + _xaResource.start(xid2, XAResource.TMNOFLAGS); } catch (XAException e) { @@ -366,7 +366,7 @@ public class TopicTest extends AbstractXATestCase { _topicConnection.start(); _logger.debug("start xid1"); - _xaResource.start(xid1, XAResource.TMSUCCESS); + _xaResource.start(xid1, XAResource.TMNOFLAGS); // start the connection _topicConnection.start(); _logger.debug("produce a message with sequence number 1"); @@ -388,7 +388,7 @@ public class TopicTest extends AbstractXATestCase try { _logger.debug("start xid2"); - _xaResource.start(xid2, XAResource.TMSUCCESS); + _xaResource.start(xid2, XAResource.TMNOFLAGS); _logger.debug("receive the previously produced message"); TextMessage message = (TextMessage) xaDurSub.receive(1000); if (message == null) @@ -430,7 +430,7 @@ public class TopicTest extends AbstractXATestCase try { _logger.debug("start xid3"); - _xaResource.start(xid3, XAResource.TMSUCCESS); + _xaResource.start(xid3, XAResource.TMNOFLAGS); _logger.debug(" receive the previously aborted consumed message"); TextMessage message = (TextMessage) xaDurSub.receive(1000); if (message == null) @@ -457,7 +457,7 @@ public class TopicTest extends AbstractXATestCase try { _logger.debug("start xid4"); - _xaResource.start(xid4, XAResource.TMSUCCESS); + _xaResource.start(xid4, XAResource.TMNOFLAGS); _logger.debug("check that topic is empty"); TextMessage message = (TextMessage) xaDurSub.receive(1000); if (message != null) @@ -543,7 +543,7 @@ public class TopicTest extends AbstractXATestCase { _logger.debug(" consume 2 messages respectively with tx1, tx2 and tx3"); //----- start xid1 - _xaResource.start(xid1, XAResource.TMSUCCESS); + _xaResource.start(xid1, XAResource.TMNOFLAGS); // receive the 2 first messages for (int i = 1; i <= 2; i++) { @@ -559,7 +559,7 @@ public class TopicTest extends AbstractXATestCase } _xaResource.end(xid1, XAResource.TMSUSPEND); //----- start xid2 - _xaResource.start(xid2, XAResource.TMSUCCESS); + _xaResource.start(xid2, XAResource.TMNOFLAGS); // receive the 2 first messages for (int i = 3; i <= 4; i++) { @@ -575,7 +575,7 @@ public class TopicTest extends AbstractXATestCase } _xaResource.end(xid2, XAResource.TMSUSPEND); //----- start xid3 - _xaResource.start(xid3, XAResource.TMSUCCESS); + _xaResource.start(xid3, XAResource.TMNOFLAGS); // receive the 2 first messages for (int i = 5; i <= 6; i++) { @@ -647,7 +647,7 @@ public class TopicTest extends AbstractXATestCase { // consume messages 1 - 4 + 7 //----- start xid1 - _xaResource.start(xid4, XAResource.TMSUCCESS); + _xaResource.start(xid4, XAResource.TMNOFLAGS); for (int i = 1; i <= 5; i++) { @@ -676,7 +676,7 @@ public class TopicTest extends AbstractXATestCase try { // start xid6 - _xaResource.start(xid6, XAResource.TMSUCCESS); + _xaResource.start(xid6, XAResource.TMNOFLAGS); // should now be empty message = (TextMessage) xaDurSub.receive(1000); if (message != null) @@ -769,7 +769,7 @@ public class TopicTest extends AbstractXATestCase { // consume 2 messages respectively with tx1, tx2 and tx3 //----- start xid1 - _xaResource.start(xid1, XAResource.TMSUCCESS); + _xaResource.start(xid1, XAResource.TMNOFLAGS); // receive the 2 first messages for (int i = 1; i <= 2; i++) { @@ -785,7 +785,7 @@ public class TopicTest extends AbstractXATestCase } _xaResource.end(xid1, XAResource.TMSUCCESS); //----- start xid2 - _xaResource.start(xid2, XAResource.TMSUCCESS); + _xaResource.start(xid2, XAResource.TMNOFLAGS); // receive the 2 first messages for (int i = 3; i <= 4; i++) { @@ -801,7 +801,7 @@ public class TopicTest extends AbstractXATestCase } _xaResource.end(xid2, XAResource.TMSUCCESS); //----- start xid3 - _xaResource.start(xid3, XAResource.TMSUCCESS); + _xaResource.start(xid3, XAResource.TMNOFLAGS); // receive the 2 first messages for (int i = 5; i <= 6; i++) { @@ -863,7 +863,7 @@ public class TopicTest extends AbstractXATestCase // xid1 has been aborted redo the job! // consume 2 messages with tx1 //----- start xid1 - _xaResource.start(xid1, XAResource.TMSUCCESS); + _xaResource.start(xid1, XAResource.TMNOFLAGS); // receive the 2 first messages for (int i = 1; i <= 2; i++) { @@ -940,7 +940,7 @@ public class TopicTest extends AbstractXATestCase { // consume messages 1 - 4 //----- start xid1 - _xaResource.start(xid4, XAResource.TMSUCCESS); + _xaResource.start(xid4, XAResource.TMNOFLAGS); for (int i = 1; i <= 4; i++) { message = (TextMessage) xaDurSub.receive(1000); @@ -955,7 +955,7 @@ public class TopicTest extends AbstractXATestCase } _xaResource.end(xid4, XAResource.TMSUSPEND); // consume messages 8 - 10 - _xaResource.start(xid5, XAResource.TMSUCCESS); + _xaResource.start(xid5, XAResource.TMNOFLAGS); for (int i = 7; i <= 10; i++) { message = (TextMessage) xaDurSub.receive(1000); @@ -1000,7 +1000,7 @@ public class TopicTest extends AbstractXATestCase try { // start xid6 - _xaResource.start(xid6, XAResource.TMSUCCESS); + _xaResource.start(xid6, XAResource.TMNOFLAGS); // should now be empty message = (TextMessage) xaDurSub.receive(1000); if (message != null) @@ -1061,7 +1061,7 @@ public class TopicTest extends AbstractXATestCase { _topicConnection.start(); //----- start xid1 - _xaResource.start(xid1, XAResource.TMSUCCESS); + _xaResource.start(xid1, XAResource.TMNOFLAGS); // start the connection _topicConnection.start(); // produce a message with sequence number 1 @@ -1083,7 +1083,7 @@ public class TopicTest extends AbstractXATestCase try { // start xid2 - _xaResource.start(xid2, XAResource.TMSUCCESS); + _xaResource.start(xid2, XAResource.TMNOFLAGS); // receive the previously produced message TextMessage message = (TextMessage) xaDurSub.receive(1000); if (message == null) @@ -1166,7 +1166,7 @@ public class TopicTest extends AbstractXATestCase try { // start xid3 - _xaResource.start(xid3, XAResource.TMSUCCESS); + _xaResource.start(xid3, XAResource.TMNOFLAGS); // receive the previously produced message and aborted TextMessage message = (TextMessage) xaDurSub.receive(1000); if (message == null) @@ -1193,7 +1193,7 @@ public class TopicTest extends AbstractXATestCase try { // start xid4 - _xaResource.start(xid4, XAResource.TMSUCCESS); + _xaResource.start(xid4, XAResource.TMNOFLAGS); // should now be empty TextMessage message = (TextMessage) xaDurSub.receive(1000); if (message != null) @@ -1248,7 +1248,7 @@ public class TopicTest extends AbstractXATestCase // start the xaResource for xid1 try { - _xaResource.start(xid1, XAResource.TMSUCCESS); + _xaResource.start(xid1, XAResource.TMNOFLAGS); } catch (XAException e) { @@ -1428,7 +1428,7 @@ public class TopicTest extends AbstractXATestCase // close the session that deactivates the durable subscriber stSession.close(); _logger.debug("migrate the durable subscriber to an xa one"); - _xaResource.start(xid1, XAResource.TMSUCCESS); + _xaResource.start(xid1, XAResource.TMNOFLAGS); durSub = _session.createDurableSubscriber(_topic, durSubName); _logger.debug(" consume the second message with that xa durable subscriber and abort it"); message = (TextMessage) durSub.receive(1000); @@ -1583,7 +1583,7 @@ public class TopicTest extends AbstractXATestCase _logger.debug(" migrate the durable subscriber to an xa one"); _session = _topicConnection.createXATopicSession(); _xaResource = _session.getXAResource(); - _xaResource.start(xid2, XAResource.TMSUCCESS); + _xaResource.start(xid2, XAResource.TMNOFLAGS); durSub = _session.createDurableSubscriber(_topic, durSubName); lock = new AtomicBoolean(); reset(); |