From commits-return-63580-archive-asf-public=cust-asf.ponee.io@activemq.apache.org Wed Jun 9 19:00:54 2021 Return-Path: X-Original-To: archive-asf-public@cust-asf.ponee.io Delivered-To: archive-asf-public@cust-asf.ponee.io Received: from mxout1-ec2-va.apache.org (mxout1-ec2-va.apache.org [3.227.148.255]) by mx-eu-01.ponee.io (Postfix) with ESMTPS id D370018063F for ; Wed, 9 Jun 2021 21:00:53 +0200 (CEST) Received: from mail.apache.org (mailroute1-lw-us.apache.org [207.244.88.153]) by mxout1-ec2-va.apache.org (ASF Mail Server at mxout1-ec2-va.apache.org) with SMTP id 14CC43EDC6 for ; Wed, 9 Jun 2021 19:00:53 +0000 (UTC) Received: (qmail 38450 invoked by uid 500); 9 Jun 2021 19:00:52 -0000 Mailing-List: contact commits-help@activemq.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@activemq.apache.org Delivered-To: mailing list commits@activemq.apache.org Received: (qmail 38441 invoked by uid 99); 9 Jun 2021 19:00:52 -0000 Received: from ec2-52-202-80-70.compute-1.amazonaws.com (HELO gitbox.apache.org) (52.202.80.70) by apache.org (qpsmtpd/0.29) with ESMTP; Wed, 09 Jun 2021 19:00:52 +0000 Received: by gitbox.apache.org (ASF Mail Server at gitbox.apache.org, from userid 33) id 6086E81A86; Wed, 9 Jun 2021 19:00:52 +0000 (UTC) Date: Wed, 09 Jun 2021 19:00:52 +0000 To: "commits@activemq.apache.org" Subject: [activemq-artemis] branch main updated: ARTEMIS-3338 Preserve prepared XA transactions on connection failure MIME-Version: 1.0 Content-Type: text/plain; charset=utf-8 Content-Transfer-Encoding: 8bit Message-ID: <162326525205.15043.1173428309438135078@gitbox.apache.org> From: clebertsuconic@apache.org X-Git-Host: gitbox.apache.org X-Git-Repo: activemq-artemis X-Git-Refname: refs/heads/main X-Git-Reftype: branch X-Git-Oldrev: 05498c350eb8be49dd26a0b593516b1988df6108 X-Git-Newrev: bafefdc8ece23801ac8fb7f5879faa9c3107ae5f X-Git-Rev: bafefdc8ece23801ac8fb7f5879faa9c3107ae5f X-Git-NotificationType: ref_changed_plus_diff X-Git-Multimail-Version: 1.5.dev Auto-Submitted: auto-generated This is an automated email from the ASF dual-hosted git repository. clebertsuconic pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git The following commit(s) were added to refs/heads/main by this push: new bafefdc ARTEMIS-3338 Preserve prepared XA transactions on connection failure bafefdc is described below commit bafefdc8ece23801ac8fb7f5879faa9c3107ae5f Author: Domenico Francesco Bruscino AuthorDate: Wed Jun 9 10:40:49 2021 +0200 ARTEMIS-3338 Preserve prepared XA transactions on connection failure --- .../core/protocol/openwire/OpenWireConnection.java | 2 +- .../core/server/impl/ServerSessionImpl.java | 6 +-- .../artemis/core/transaction/Transaction.java | 2 +- .../core/transaction/impl/TransactionImpl.java | 6 ++- .../tests/integration/xa/SessionFailureXATest.java | 63 ++++++++++++++-------- .../core/postoffice/impl/BindingsImplTest.java | 4 +- 6 files changed, 50 insertions(+), 33 deletions(-) diff --git a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java index 8807c7a..5b4586d 100644 --- a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java +++ b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java @@ -687,7 +687,7 @@ public class OpenWireConnection extends AbstractRemotingConnection implements Se public void fail(ActiveMQException me, String message) { for (Transaction tx : txMap.values()) { - tx.rollbackIfPossible(); + tx.tryRollback(); } if (me != null) { diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java index 12a080e..542a974 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java @@ -400,19 +400,17 @@ public class ServerSessionImpl implements ServerSession, FailureListener { Transaction txToRollback = tx; if (txToRollback != null) { - if (txToRollback.getXid() != null) { + if (txToRollback.tryRollback() && txToRollback.getXid() != null) { resourceManager.removeTransaction(txToRollback.getXid(), remotingConnection); } - txToRollback.rollbackIfPossible(); } txToRollback = pendingTX; if (txToRollback != null) { - if (txToRollback.getXid() != null) { + if (txToRollback.tryRollback() && txToRollback.getXid() != null) { resourceManager.removeTransaction(txToRollback.getXid(), remotingConnection); } - txToRollback.rollbackIfPossible(); } } else { diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/Transaction.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/Transaction.java index 6a6be0a..41aa7ed 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/Transaction.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/Transaction.java @@ -54,7 +54,7 @@ public interface Transaction { /** In a ServerSession failure scenario,\ * we may try to rollback, however only if it's not prepared. * In case it's prepared, we will just let it be and let the transaction manager to deal with it */ - void rollbackIfPossible(); + boolean tryRollback(); long getID(); diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImpl.java index d35c2a8..1b0d660 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImpl.java @@ -341,16 +341,17 @@ public class TransactionImpl implements Transaction { } @Override - public void rollbackIfPossible() { + public boolean tryRollback() { synchronized (timeoutLock) { if (state == State.ROLLEDBACK) { // I don't think this could happen, but just in case logger.debug("TransactionImpl::rollbackIfPossible::" + this + " is being ignored"); - return; + return true; } if (state != State.PREPARED) { try { internalRollback(); + return true; } catch (Exception e) { // nothing we can do beyond logging // no need to special handler here as this was not even supposed to happen at this point @@ -359,6 +360,7 @@ public class TransactionImpl implements Transaction { } } } + return false; } @Override diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/xa/SessionFailureXATest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/xa/SessionFailureXATest.java index 2e51087..4f1c474 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/xa/SessionFailureXATest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/xa/SessionFailureXATest.java @@ -110,15 +110,20 @@ public class SessionFailureXATest extends ActiveMQTestBase { @Test public void testFailureWithXAEnd() throws Exception { - testFailure(true); + testFailure(true, false); } @Test public void testFailureWithoutXAEnd() throws Exception { - testFailure(false); + testFailure(false, false); } - public void testFailure(boolean xaEnd) throws Exception { + @Test + public void testFailureWithXAPrepare() throws Exception { + testFailure(true, true); + } + + public void testFailure(boolean xaEnd, boolean xaPrepare) throws Exception { ClientSession clientSession2 = sessionFactory.createSession(false, true, true); try { @@ -160,6 +165,10 @@ public class SessionFailureXATest extends ActiveMQTestBase { // We are validating both cases, where xaEnd succeeded and didn't succeed // so this tests is parameterized to validate both cases. clientSession.end(xid, XAResource.TMSUCCESS); + + if (xaPrepare) { + clientSession.prepare(xid); + } } Wait.assertEquals(1, () -> messagingService.getSessions().size()); @@ -171,7 +180,11 @@ public class SessionFailureXATest extends ActiveMQTestBase { Wait.assertEquals(0, () -> messagingService.getSessions().size()); - Wait.assertEquals(0, messagingService.getResourceManager()::size); + if (xaPrepare) { + Wait.assertEquals(1, messagingService.getResourceManager()::size); + } else { + Wait.assertEquals(0, messagingService.getResourceManager()::size); + } locator = createInVMNonHALocator(); sessionFactory = createSessionFactory(locator); @@ -188,25 +201,29 @@ public class SessionFailureXATest extends ActiveMQTestBase { HashSet bodies = new HashSet<>(); m = clientConsumer.receive(1000); - Assert.assertNotNull(m); - m.acknowledge(); - assertOrTrack(xaEnd, m, bodies, "m1"); - m = clientConsumer.receive(1000); - Assert.assertNotNull(m); - m.acknowledge(); - assertOrTrack(xaEnd, m, bodies, "m2"); - m = clientConsumer.receive(1000); - Assert.assertNotNull(m); - m.acknowledge(); - assertOrTrack(xaEnd, m, bodies, "m3"); - m = clientConsumer.receive(1000); - Assert.assertNotNull(m); - m.acknowledge(); - assertOrTrack(xaEnd, m, bodies, "m4"); - - if (!xaEnd) { - // order is not guaranteed b/c the m4 async ack may not have been processed when there is no sync end call - assertEquals("got all bodies", 4, bodies.size()); + if (xaPrepare) { + Assert.assertNull(m); + } else { + Assert.assertNotNull(m); + m.acknowledge(); + assertOrTrack(xaEnd, m, bodies, "m1"); + m = clientConsumer.receive(1000); + Assert.assertNotNull(m); + m.acknowledge(); + assertOrTrack(xaEnd, m, bodies, "m2"); + m = clientConsumer.receive(1000); + Assert.assertNotNull(m); + m.acknowledge(); + assertOrTrack(xaEnd, m, bodies, "m3"); + m = clientConsumer.receive(1000); + Assert.assertNotNull(m); + m.acknowledge(); + assertOrTrack(xaEnd, m, bodies, "m4"); + + if (!xaEnd) { + // order is not guaranteed b/c the m4 async ack may not have been processed when there is no sync end call + assertEquals("got all bodies", 4, bodies.size()); + } } } diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/BindingsImplTest.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/BindingsImplTest.java index 80cd76a..72eb506 100644 --- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/BindingsImplTest.java +++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/BindingsImplTest.java @@ -140,8 +140,8 @@ public class BindingsImplTest extends ActiveMQTestBase { } @Override - public void rollbackIfPossible() { - + public boolean tryRollback() { + return true; } @Override