[tx-control] Add support for resource recovery when using XA
git-svn-id: https://svn.apache.org/repos/asf/aries/trunk@1750850 13f79535-47bb-0310-9956-ffa450edef68
diff --git a/tx-control/pom.xml b/tx-control/pom.xml
index cdac4cb..28ccb0e 100644
--- a/tx-control/pom.xml
+++ b/tx-control/pom.xml
@@ -104,7 +104,7 @@
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
- <version>1.6.6</version>
+ <version>1.7.0</version>
</dependency>
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
diff --git a/tx-control/tx-control-api/pom.xml b/tx-control/tx-control-api/pom.xml
index 3ac65b7..3ca828c 100644
--- a/tx-control/tx-control-api/pom.xml
+++ b/tx-control/tx-control-api/pom.xml
@@ -56,4 +56,5 @@
</plugin>
</plugins>
</build>
+
</project>
\ No newline at end of file
diff --git a/tx-control/tx-control-api/src/main/java/org/osgi/service/transaction/control/TransactionContext.java b/tx-control/tx-control-api/src/main/java/org/osgi/service/transaction/control/TransactionContext.java
index 6a96ec4..ddb9842 100644
--- a/tx-control/tx-control-api/src/main/java/org/osgi/service/transaction/control/TransactionContext.java
+++ b/tx-control/tx-control-api/src/main/java/org/osgi/service/transaction/control/TransactionContext.java
@@ -19,6 +19,8 @@
import javax.transaction.xa.XAResource;
+import org.osgi.service.transaction.control.recovery.RecoverableXAResource;
+
/**
* A transaction context defines the current transaction, and allows resources
* to register information and/or synchronisations
@@ -110,10 +112,14 @@
* Register an XA resource with the current transaction
*
* @param resource
+ * @param name The resource name used for recovery, may be <code>null</code>
+ * if this resource is not recoverable. If a name is passed then
+ * a corresponding {@link RecoverableXAResource} must be registered
+ * in the service registry
* @throws IllegalStateException if no transaction is active, or the current
* transaction is not XA capable
*/
- void registerXAResource(XAResource resource) throws IllegalStateException;
+ void registerXAResource(XAResource resource, String name) throws IllegalStateException;
/**
* Register an XA resource with the current transaction
diff --git a/tx-control/tx-control-api/src/main/java/org/osgi/service/transaction/control/recovery/RecoverableXAResource.java b/tx-control/tx-control-api/src/main/java/org/osgi/service/transaction/control/recovery/RecoverableXAResource.java
new file mode 100644
index 0000000..bb0fa8b
--- /dev/null
+++ b/tx-control/tx-control-api/src/main/java/org/osgi/service/transaction/control/recovery/RecoverableXAResource.java
@@ -0,0 +1,75 @@
+/*
+ * Copyright (c) OSGi Alliance (2016). All Rights Reserved.
+ *
+ * Licensed 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.
+ */
+package org.osgi.service.transaction.control.recovery;
+
+import javax.transaction.xa.XAResource;
+
+import org.osgi.service.transaction.control.ResourceProvider;
+import org.osgi.service.transaction.control.TransactionContext;
+
+/**
+ * A {@link RecoverableXAResource} service may be provided by a
+ * {@link ResourceProvider} if they are able to support XA recovery
+ * operations.
+ *
+ * There are two main sorts of recovery:
+ *
+ * <ul>
+ * <li>Recovery after a remote failure, where the local transaction
+ * manager runs throughout</li>
+ * <li>Recovery after a local failure, where the transaction manager
+ * replays in-doubt transactions from its log</li>
+ * </ul>
+ *
+ * This service is used in both of these cases.
+ *
+ * The identifier returned by {@link #getId()} provides a persistent name
+ * that can be used to correlate usage of the resource both before and after
+ * failure. This identifier must also be passed to
+ * {@link TransactionContext#registerXAResource(XAResource, String)} each time
+ * the recoverable resource is used.
+ *
+ */
+public interface RecoverableXAResource {
+
+ /**
+ * Get the id of this resource. This should be unique, and persist between restarts
+ * @return an identifier, never <code>null</code>
+ */
+ String getId();
+
+ /**
+ * Get a new, valid XAResource that can be used in recovery
+ *
+ * This XAResource will be returned later using the
+ * {@link #releaseXAResource(XAResource)} method
+ *
+ * @return a valid, connected, XAResource
+ *
+ * @throws Exception If it is not possible to acquire a valid
+ * XAResource at the current time, for example if the database
+ * is temporarily unavailable.
+ */
+ XAResource getXAResource() throws Exception;
+
+ /**
+ * Release the XAResource that has been used for recovery
+ *
+ * @param xaRes An {@link XAResource} previously returned
+ * by {@link #getXAResource()}
+ */
+ void releaseXAResource(XAResource xaRes);
+}
diff --git a/tx-control/tx-control-api/src/main/java/org/osgi/service/transaction/control/recovery/TransactionRecovery.java b/tx-control/tx-control-api/src/main/java/org/osgi/service/transaction/control/recovery/TransactionRecovery.java
deleted file mode 100644
index b993f7c..0000000
--- a/tx-control/tx-control-api/src/main/java/org/osgi/service/transaction/control/recovery/TransactionRecovery.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Copyright (c) OSGi Alliance (2016). All Rights Reserved.
- *
- * Licensed 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.
- */
-package org.osgi.service.transaction.control.recovery;
-
-import javax.transaction.xa.XAResource;
-
-/**
- * This service interface is published by Transaction control services that are
- * able to support recovery. Any recoverable resources should register
- * themselves with all available recovery services as they are created.
- */
-public interface TransactionRecovery {
-
- /**
- * Allow the {@link TransactionRecovery} service to attempt to recover any
- * incomplete XA transactions. Any recovery failures that occur must be
- * logged and not thrown to the caller of this service.
- *
- * @param resource
- */
- public void recover(XAResource resource);
-
-}
diff --git a/tx-control/tx-control-itests/src/test/java/org/apache/aries/tx/control/itests/XATransactionTest.java b/tx-control/tx-control-itests/src/test/java/org/apache/aries/tx/control/itests/XATransactionTest.java
index e22708d..e2b075e 100644
--- a/tx-control/tx-control-itests/src/test/java/org/apache/aries/tx/control/itests/XATransactionTest.java
+++ b/tx-control/tx-control-itests/src/test/java/org/apache/aries/tx/control/itests/XATransactionTest.java
@@ -182,7 +182,7 @@
return null;
});
- txControl.getCurrentContext().registerXAResource(new PoisonResource());
+ txControl.getCurrentContext().registerXAResource(new PoisonResource(), null);
return null;
});
diff --git a/tx-control/tx-control-jpa-itests/src/test/java/org/apache/aries/tx/control/itests/XAJPATransactionTest.java b/tx-control/tx-control-jpa-itests/src/test/java/org/apache/aries/tx/control/itests/XAJPATransactionTest.java
index 43ecd56..9e705dd 100644
--- a/tx-control/tx-control-jpa-itests/src/test/java/org/apache/aries/tx/control/itests/XAJPATransactionTest.java
+++ b/tx-control/tx-control-jpa-itests/src/test/java/org/apache/aries/tx/control/itests/XAJPATransactionTest.java
@@ -325,7 +325,7 @@
return null;
});
- txControl.getCurrentContext().registerXAResource(new PoisonResource());
+ txControl.getCurrentContext().registerXAResource(new PoisonResource(), null);
return null;
});
diff --git a/tx-control/tx-control-provider-jdbc-xa/src/main/java/org/apache/aries/tx/control/jdbc/xa/impl/XAEnabledTxContextBindingConnection.java b/tx-control/tx-control-provider-jdbc-xa/src/main/java/org/apache/aries/tx/control/jdbc/xa/impl/XAEnabledTxContextBindingConnection.java
index 73d330d4..764bb0c 100644
--- a/tx-control/tx-control-provider-jdbc-xa/src/main/java/org/apache/aries/tx/control/jdbc/xa/impl/XAEnabledTxContextBindingConnection.java
+++ b/tx-control/tx-control-provider-jdbc-xa/src/main/java/org/apache/aries/tx/control/jdbc/xa/impl/XAEnabledTxContextBindingConnection.java
@@ -79,7 +79,7 @@
} else if (txContext.supportsXA() && xaEnabled) {
toClose = dataSource.getConnection();
toReturn = new TxConnectionWrapper(toClose);
- txContext.registerXAResource(getXAResource(toClose));
+ txContext.registerXAResource(getXAResource(toClose), null);
} else if (txContext.supportsLocal() && localEnabled) {
toClose = dataSource.getConnection();
toReturn = new TxConnectionWrapper(toClose);
diff --git a/tx-control/tx-control-provider-jdbc-xa/src/test/java/org/apache/aries/tx/control/jdbc/xa/impl/XAEnabledTxContextBindingConnectionTest.java b/tx-control/tx-control-provider-jdbc-xa/src/test/java/org/apache/aries/tx/control/jdbc/xa/impl/XAEnabledTxContextBindingConnectionTest.java
index 9bcf955..d447d79 100644
--- a/tx-control/tx-control-provider-jdbc-xa/src/test/java/org/apache/aries/tx/control/jdbc/xa/impl/XAEnabledTxContextBindingConnectionTest.java
+++ b/tx-control/tx-control-provider-jdbc-xa/src/test/java/org/apache/aries/tx/control/jdbc/xa/impl/XAEnabledTxContextBindingConnectionTest.java
@@ -211,7 +211,7 @@
Mockito.verify(rawConnection, times(2)).isValid(500);
- Mockito.verify(context).registerXAResource(xaResource);
+ Mockito.verify(context).registerXAResource(xaResource, null);
Mockito.verify(context).postCompletion(Mockito.any());
@@ -226,7 +226,7 @@
xaConn.isValid(500);
Mockito.verify(rawConnection, times(2)).isValid(500);
- Mockito.verify(context).registerXAResource(xaResource);
+ Mockito.verify(context).registerXAResource(xaResource, null);
Mockito.verify(context).postCompletion(Mockito.any());
diff --git a/tx-control/tx-control-provider-jpa-xa/src/main/java/org/apache/aries/tx/control/jpa/xa/eclipse/impl/EclipseTxControlPlatform.java b/tx-control/tx-control-provider-jpa-xa/src/main/java/org/apache/aries/tx/control/jpa/xa/eclipse/impl/EclipseTxControlPlatform.java
index 0b6cbd2..bfa2b77 100644
--- a/tx-control/tx-control-provider-jpa-xa/src/main/java/org/apache/aries/tx/control/jpa/xa/eclipse/impl/EclipseTxControlPlatform.java
+++ b/tx-control/tx-control-provider-jpa-xa/src/main/java/org/apache/aries/tx/control/jpa/xa/eclipse/impl/EclipseTxControlPlatform.java
@@ -204,7 +204,7 @@
@Override
public boolean enlistResource(XAResource xaRes)
throws IllegalStateException, RollbackException, SystemException {
- context.registerXAResource(xaRes);
+ context.registerXAResource(xaRes, null);
return true;
}
diff --git a/tx-control/tx-control-provider-jpa-xa/src/main/java/org/apache/aries/tx/control/jpa/xa/impl/JPAEntityManagerProviderFactoryImpl.java b/tx-control/tx-control-provider-jpa-xa/src/main/java/org/apache/aries/tx/control/jpa/xa/impl/JPAEntityManagerProviderFactoryImpl.java
index 7b4c7fe..c8b132c 100644
--- a/tx-control/tx-control-provider-jpa-xa/src/main/java/org/apache/aries/tx/control/jpa/xa/impl/JPAEntityManagerProviderFactoryImpl.java
+++ b/tx-control/tx-control-provider-jpa-xa/src/main/java/org/apache/aries/tx/control/jpa/xa/impl/JPAEntityManagerProviderFactoryImpl.java
@@ -210,7 +210,7 @@
toReturn = new ScopedConnectionWrapper(toClose);
} else if (txContext.supportsXA()) {
toReturn = new TxConnectionWrapper(toClose);
- txContext.registerXAResource(getXAResource(toClose));
+ txContext.registerXAResource(getXAResource(toClose), null);
} else {
throw new TransactionException(
"There is a transaction active, but it does not support XA participants");
diff --git a/tx-control/tx-control-provider-jpa-xa/src/main/java/org/apache/aries/tx/control/jpa/xa/openjpa/impl/OpenJPATxControlPlatform.java b/tx-control/tx-control-provider-jpa-xa/src/main/java/org/apache/aries/tx/control/jpa/xa/openjpa/impl/OpenJPATxControlPlatform.java
index 60e22c5..7219613 100644
--- a/tx-control/tx-control-provider-jpa-xa/src/main/java/org/apache/aries/tx/control/jpa/xa/openjpa/impl/OpenJPATxControlPlatform.java
+++ b/tx-control/tx-control-provider-jpa-xa/src/main/java/org/apache/aries/tx/control/jpa/xa/openjpa/impl/OpenJPATxControlPlatform.java
@@ -132,7 +132,7 @@
@Override
public boolean enlistResource(XAResource xaRes) throws IllegalStateException, RollbackException, SystemException {
- txControl.getCurrentContext().registerXAResource(xaRes);
+ txControl.getCurrentContext().registerXAResource(xaRes, null);
return true;
}
diff --git a/tx-control/tx-control-provider-jpa-xa/src/test/java/org/apache/aries/tx/control/jpa/xa/impl/XATxContextBindingEntityManagerTest.java b/tx-control/tx-control-provider-jpa-xa/src/test/java/org/apache/aries/tx/control/jpa/xa/impl/XATxContextBindingEntityManagerTest.java
index 4fbd9e7..12c745c 100644
--- a/tx-control/tx-control-provider-jpa-xa/src/test/java/org/apache/aries/tx/control/jpa/xa/impl/XATxContextBindingEntityManagerTest.java
+++ b/tx-control/tx-control-provider-jpa-xa/src/test/java/org/apache/aries/tx/control/jpa/xa/impl/XATxContextBindingEntityManagerTest.java
@@ -108,7 +108,7 @@
Mockito.verify(rawEm, times(2)).isOpen();
Mockito.verify(rawEm, times(0)).getTransaction();
- Mockito.verify(context, times(0)).registerXAResource(Mockito.any());
+ Mockito.verify(context, times(0)).registerXAResource(Mockito.any(), Mockito.anyString());
Mockito.verify(context).postCompletion(Mockito.any());
}
diff --git a/tx-control/tx-control-service-common/src/main/java/org/apache/aries/tx/control/service/common/impl/NoTransactionContextImpl.java b/tx-control/tx-control-service-common/src/main/java/org/apache/aries/tx/control/service/common/impl/NoTransactionContextImpl.java
index 555d860..aa3d6d0 100644
--- a/tx-control/tx-control-service-common/src/main/java/org/apache/aries/tx/control/service/common/impl/NoTransactionContextImpl.java
+++ b/tx-control/tx-control-service-common/src/main/java/org/apache/aries/tx/control/service/common/impl/NoTransactionContextImpl.java
@@ -80,7 +80,7 @@
}
@Override
- public void registerXAResource(XAResource resource) {
+ public void registerXAResource(XAResource resource, String recoveryName) {
throw new IllegalStateException("No transaction is active");
}
diff --git a/tx-control/tx-control-service-common/src/test/java/org/apache/aries/tx/control/service/common/impl/NoTransactionContextTest.java b/tx-control/tx-control-service-common/src/test/java/org/apache/aries/tx/control/service/common/impl/NoTransactionContextTest.java
index ae3981d..9426bb5 100644
--- a/tx-control/tx-control-service-common/src/test/java/org/apache/aries/tx/control/service/common/impl/NoTransactionContextTest.java
+++ b/tx-control/tx-control-service-common/src/test/java/org/apache/aries/tx/control/service/common/impl/NoTransactionContextTest.java
@@ -87,7 +87,7 @@
@Test(expected=IllegalStateException.class)
public void testXAResourceRegistration() {
- ctx.registerXAResource(xaResource);
+ ctx.registerXAResource(xaResource, null);
}
@Test
diff --git a/tx-control/tx-control-service-local/src/main/java/org/apache/aries/tx/control/service/local/impl/TransactionContextImpl.java b/tx-control/tx-control-service-local/src/main/java/org/apache/aries/tx/control/service/local/impl/TransactionContextImpl.java
index bf13d8e..1c6c14e 100644
--- a/tx-control/tx-control-service-local/src/main/java/org/apache/aries/tx/control/service/local/impl/TransactionContextImpl.java
+++ b/tx-control/tx-control-service-local/src/main/java/org/apache/aries/tx/control/service/local/impl/TransactionContextImpl.java
@@ -133,7 +133,7 @@
}
@Override
- public void registerXAResource(XAResource resource) {
+ public void registerXAResource(XAResource resource, String name) {
throw new IllegalStateException("Not an XA manager");
}
diff --git a/tx-control/tx-control-service-local/src/test/java/org/apache/aries/tx/control/service/local/impl/TransactionContextTest.java b/tx-control/tx-control-service-local/src/test/java/org/apache/aries/tx/control/service/local/impl/TransactionContextTest.java
index 9de870e..634bfc9 100644
--- a/tx-control/tx-control-service-local/src/test/java/org/apache/aries/tx/control/service/local/impl/TransactionContextTest.java
+++ b/tx-control/tx-control-service-local/src/test/java/org/apache/aries/tx/control/service/local/impl/TransactionContextTest.java
@@ -110,7 +110,7 @@
@Test(expected=IllegalStateException.class)
public void testXAResourceRegistration() {
- ctx.registerXAResource(xaResource);
+ ctx.registerXAResource(xaResource, null);
}
@Test
diff --git a/tx-control/tx-control-service-xa/bnd.bnd b/tx-control/tx-control-service-xa/bnd.bnd
index 812fb63..7e61eea 100644
--- a/tx-control/tx-control-service-xa/bnd.bnd
+++ b/tx-control/tx-control-service-xa/bnd.bnd
@@ -3,7 +3,8 @@
# Export the API so that this is an easily deployable bundle
-Export-Package: org.osgi.service.transaction.control
+Export-Package: org.osgi.service.transaction.control,\
+ org.osgi.service.transaction.control.recovery
# This bundle repackages code from a variety of places to make the
diff --git a/tx-control/tx-control-service-xa/src/main/java/org/apache/aries/tx/control/service/xa/impl/Config.java b/tx-control/tx-control-service-xa/src/main/java/org/apache/aries/tx/control/service/xa/impl/Config.java
index 4039744..3cf2d7f 100644
--- a/tx-control/tx-control-service-xa/src/main/java/org/apache/aries/tx/control/service/xa/impl/Config.java
+++ b/tx-control/tx-control-service-xa/src/main/java/org/apache/aries/tx/control/service/xa/impl/Config.java
@@ -11,8 +11,8 @@
*/
@ObjectClassDefinition(pid=Activator.PID, description="Apache Aries Transaction Control Service (XA)")
@interface Config {
- @AttributeDefinition(name="Enable recovery", required=false, description="Enable recovery")
- boolean recovery_enabled() default false;
+ @AttributeDefinition(name="Enable recovery logging", required=false, description="Enable recovery logging")
+ boolean recovery_log_enabled() default false;
@AttributeDefinition(name="Recovery Log storage folder", required=false, description="Transaction Recovery Log directory")
boolean recovery_log_dir();
diff --git a/tx-control/tx-control-service-xa/src/main/java/org/apache/aries/tx/control/service/xa/impl/NamedXAResourceImpl.java b/tx-control/tx-control-service-xa/src/main/java/org/apache/aries/tx/control/service/xa/impl/NamedXAResourceImpl.java
new file mode 100644
index 0000000..d394237
--- /dev/null
+++ b/tx-control/tx-control-service-xa/src/main/java/org/apache/aries/tx/control/service/xa/impl/NamedXAResourceImpl.java
@@ -0,0 +1,143 @@
+package org.apache.aries.tx.control.service.xa.impl;
+
+import javax.transaction.xa.XAException;
+import javax.transaction.xa.XAResource;
+import javax.transaction.xa.Xid;
+
+import org.apache.geronimo.transaction.manager.NamedXAResource;
+import org.apache.geronimo.transaction.manager.RecoveryWorkAroundTransactionManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NamedXAResourceImpl implements NamedXAResource, AutoCloseable {
+
+ final Logger logger = LoggerFactory.getLogger(NamedXAResourceImpl.class);
+
+ final String name;
+ final XAResource xaResource;
+ final RecoveryWorkAroundTransactionManager transactionManager;
+ final boolean original;
+
+ boolean closed;
+
+ public NamedXAResourceImpl(String name, XAResource xaResource,
+ RecoveryWorkAroundTransactionManager transactionManager, boolean original) {
+ this.name = name;
+ this.xaResource = xaResource;
+ this.transactionManager = transactionManager;
+ this.original = original;
+ }
+
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ }
+
+ private interface XAAction {
+ void perform() throws XAException;
+ }
+
+ private interface XAReturnAction<T> {
+ T perform() throws XAException;
+ }
+
+ private void safeCall(XAAction action) throws XAException {
+ checkOpen();
+
+ try {
+ action.perform();
+ } catch (Exception e) {
+ throw handleException(e);
+ }
+ }
+
+ private <T> T safeCall(XAReturnAction<T> action) throws XAException {
+ checkOpen();
+ try {
+ return action.perform();
+ } catch (Exception e) {
+ throw handleException(e);
+ }
+ }
+
+ private void checkOpen() throws XAException {
+ if(closed) {
+ XAException xaException = new XAException("This instance of the resource named " + name + " is no longer available");
+ xaException.errorCode = XAException.XAER_RMFAIL;
+ throw xaException;
+ }
+ }
+
+ private XAException handleException(Exception e) throws XAException {
+ if(e instanceof XAException) {
+ XAException xae = (XAException) e;
+ if(xae.errorCode == 0) {
+ if(original) {
+ // We are the originally enlisted resource, and will play some tricks to attempt recovery
+ if(transactionManager.getNamedResource(name) == null) {
+ logger.error("The XA resource named {} threw an XAException but did not set the error code. There is also no RecoverableXAResource available with the name {}. It is not possible to recover from this situation and so the transaction will have to be resolved by an operator.", name, name, xae);
+ xae.errorCode = XAException.XAER_RMERR;
+ } else {
+ logger.warn("The XA resource named {} threw an XAException but did not set the error code. Changing it to be an \"RM_FAIL\" to permit recovery attempts", name, xae);
+ xae.errorCode = XAException.XAER_RMFAIL;
+ }
+ } else {
+ logger.warn("The XA resource named {} threw an XAException but did not set the error code. Recovery has already been attempted for this resource and it has not been possible to recover from this situation. The transaction will have to be resolved by an operator.", name, xae);
+ xae.errorCode = XAException.XAER_RMERR;
+ }
+ }
+ return xae;
+ } else {
+ logger.warn("The recoverable XA resource named {} threw an Exception which is not permitted by the interface. Changing it to be a \"Resource Manager Error\" XAException which prevents recovery", name, e);
+ XAException xaException = new XAException(XAException.XAER_RMERR);
+ xaException.initCause(e);
+ return xaException;
+ }
+ }
+
+ public void commit(Xid xid, boolean onePhase) throws XAException {
+ safeCall(() -> xaResource.commit(xid, onePhase));
+ }
+
+ public void end(Xid xid, int flags) throws XAException {
+ safeCall(() -> xaResource.end(xid, flags));
+ }
+
+ public void forget(Xid xid) throws XAException {
+ safeCall(() -> xaResource.forget(xid));
+ }
+
+ public int getTransactionTimeout() throws XAException {
+ return safeCall(() -> xaResource.getTransactionTimeout());
+ }
+
+ public boolean isSameRM(XAResource xares) throws XAException {
+ return safeCall(() -> xaResource.isSameRM(xares));
+ }
+
+ public int prepare(Xid xid) throws XAException {
+ return safeCall(() -> xaResource.prepare(xid));
+ }
+
+ public Xid[] recover(int flag) throws XAException {
+ return safeCall(() -> xaResource.recover(flag));
+ }
+
+ public void rollback(Xid xid) throws XAException {
+ safeCall(() -> xaResource.rollback(xid));
+ }
+
+ public boolean setTransactionTimeout(int seconds) throws XAException {
+ return safeCall(() -> xaResource.setTransactionTimeout(seconds));
+ }
+
+ public void start(Xid xid, int flags) throws XAException {
+ safeCall(() -> xaResource.start(xid, flags));
+ }
+
+}
diff --git a/tx-control/tx-control-service-xa/src/main/java/org/apache/aries/tx/control/service/xa/impl/TransactionContextImpl.java b/tx-control/tx-control-service-xa/src/main/java/org/apache/aries/tx/control/service/xa/impl/TransactionContextImpl.java
index 81305b0..67eff17 100644
--- a/tx-control/tx-control-service-xa/src/main/java/org/apache/aries/tx/control/service/xa/impl/TransactionContextImpl.java
+++ b/tx-control/tx-control-service-xa/src/main/java/org/apache/aries/tx/control/service/xa/impl/TransactionContextImpl.java
@@ -47,7 +47,7 @@
import javax.transaction.xa.Xid;
import org.apache.aries.tx.control.service.common.impl.AbstractTransactionContextImpl;
-import org.apache.geronimo.transaction.manager.GeronimoTransactionManager;
+import org.apache.geronimo.transaction.manager.RecoveryWorkAroundTransactionManager;
import org.osgi.service.transaction.control.LocalResource;
import org.osgi.service.transaction.control.TransactionContext;
import org.osgi.service.transaction.control.TransactionException;
@@ -63,7 +63,7 @@
private final AtomicReference<TransactionStatus> completionState = new AtomicReference<>();
- private final GeronimoTransactionManager transactionManager;
+ private final RecoveryWorkAroundTransactionManager transactionManager;
private final Object key;
@@ -71,7 +71,7 @@
private LocalResourceSupport localResourceSupport;
- public TransactionContextImpl(GeronimoTransactionManager transactionManager,
+ public TransactionContextImpl(RecoveryWorkAroundTransactionManager transactionManager,
boolean readOnly, LocalResourceSupport localResourceSupport) {
this.transactionManager = transactionManager;
this.readOnly = readOnly;
@@ -214,13 +214,19 @@
}
@Override
- public void registerXAResource(XAResource resource) {
+ public void registerXAResource(XAResource resource, String name) {
TransactionStatus status = getTransactionStatus();
if (status.compareTo(MARKED_ROLLBACK) > 0) {
throw new IllegalStateException("The current transaction is in state " + status);
}
try {
- currentTransaction.enlistResource(resource);
+ if(name == null) {
+ currentTransaction.enlistResource(resource);
+ } else {
+ NamedXAResourceImpl res = new NamedXAResourceImpl(name, resource, transactionManager, true);
+ postCompletion(x -> res.close());
+ currentTransaction.enlistResource(res);
+ }
} catch (Exception e) {
throw new TransactionException("The transaction was unable to enlist a resource", e);
}
@@ -288,52 +294,22 @@
}
}
}
-
- TxListener listener;
- boolean manualCallListener;
- if(!preCompletion.isEmpty() || !postCompletion.isEmpty()) {
- listener = new TxListener();
- try {
- transactionManager.registerInterposedSynchronization(listener);
- manualCallListener = false;
- } catch (Exception e) {
- manualCallListener = true;
- recordFailure(e);
- safeSetRollbackOnly();
- }
- } else {
- listener = null;
- manualCallListener = false;
- }
-
try {
- int status;
+ TxListener listener = new TxListener();
try {
+ transactionManager.registerInterposedSynchronization(listener);
+
if (getRollbackOnly()) {
// GERONIMO-4449 says that we get no beforeCompletion
// callback for rollback :(
- if(listener != null) {
- listener.beforeCompletion();
- }
+ listener.beforeCompletion();
transactionManager.rollback();
- status = Status.STATUS_ROLLEDBACK;
- completionState.set(ROLLED_BACK);
} else {
- if(manualCallListener) {
- listener.beforeCompletion();
- }
transactionManager.commit();
- status = Status.STATUS_COMMITTED;
- completionState.set(COMMITTED);
}
} catch (Exception e) {
recordFailure(e);
- status = Status.STATUS_ROLLEDBACK;
- completionState.set(ROLLED_BACK);
- }
- if(manualCallListener) {
- listener.afterCompletion(status);
}
} finally {
try {
@@ -465,7 +441,7 @@
}
private class TxListener implements Synchronization {
-
+
@Override
public void beforeCompletion() {
TransactionContextImpl.this.beforeCompletion(() -> safeSetRollbackOnly());
@@ -473,8 +449,9 @@
@Override
public void afterCompletion(int status) {
- TransactionContextImpl.this.afterCompletion(status == Status.STATUS_COMMITTED ? COMMITTED : ROLLED_BACK);
+ TransactionStatus ts = status == Status.STATUS_COMMITTED ? COMMITTED : ROLLED_BACK;
+ completionState.set(ts);
+ TransactionContextImpl.this.afterCompletion(ts);
}
-
}
}
diff --git a/tx-control/tx-control-service-xa/src/main/java/org/apache/aries/tx/control/service/xa/impl/TransactionControlImpl.java b/tx-control/tx-control-service-xa/src/main/java/org/apache/aries/tx/control/service/xa/impl/TransactionControlImpl.java
index 6d6ecbb..218f977 100644
--- a/tx-control/tx-control-service-xa/src/main/java/org/apache/aries/tx/control/service/xa/impl/TransactionControlImpl.java
+++ b/tx-control/tx-control-service-xa/src/main/java/org/apache/aries/tx/control/service/xa/impl/TransactionControlImpl.java
@@ -28,23 +28,37 @@
import java.util.Hashtable;
import java.util.Map;
+import javax.resource.spi.IllegalStateException;
+import javax.transaction.SystemException;
+import javax.transaction.xa.XAResource;
+
import org.apache.aries.tx.control.service.common.impl.AbstractTransactionContextImpl;
import org.apache.aries.tx.control.service.common.impl.AbstractTransactionControlImpl;
import org.apache.aries.tx.control.service.xa.impl.Activator.ChangeType;
import org.apache.geronimo.transaction.log.HOWLLog;
-import org.apache.geronimo.transaction.manager.GeronimoTransactionManager;
+import org.apache.geronimo.transaction.manager.NamedXAResource;
+import org.apache.geronimo.transaction.manager.NamedXAResourceFactory;
+import org.apache.geronimo.transaction.manager.RecoveryWorkAroundTransactionManager;
import org.apache.geronimo.transaction.manager.XidFactory;
import org.apache.geronimo.transaction.manager.XidFactoryImpl;
import org.osgi.framework.BundleContext;
import org.osgi.framework.Constants;
+import org.osgi.framework.ServiceReference;
+import org.osgi.service.transaction.control.recovery.RecoverableXAResource;
+import org.osgi.util.tracker.ServiceTracker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class TransactionControlImpl extends AbstractTransactionControlImpl {
+ private static final Logger logger = LoggerFactory.getLogger(TransactionControlImpl.class);
+
private Map<String, Object> config;
private final XidFactory xidFactory;
private final HOWLLog log;
- private final GeronimoTransactionManager transactionManager;
+ private final RecoveryWorkAroundTransactionManager transactionManager;
private final LocalResourceSupport localResourceSupport;
+ private final ServiceTracker<RecoverableXAResource, RecoverableXAResource> recoverableResources;
public TransactionControlImpl(BundleContext ctx, Map<String, Object> config) throws Exception {
@@ -58,8 +72,72 @@
log.doStart();
}
- transactionManager = new GeronimoTransactionManager(getTimeout(),
+ transactionManager = new RecoveryWorkAroundTransactionManager(getTimeout(),
xidFactory, log);
+
+ if(log != null) {
+ recoverableResources =
+ new ServiceTracker<RecoverableXAResource, RecoverableXAResource>(
+ ctx, RecoverableXAResource.class, null) {
+
+ @Override
+ public RecoverableXAResource addingService(
+ ServiceReference<RecoverableXAResource> reference) {
+ RecoverableXAResource resource = super.addingService(reference);
+
+ if(resource.getId() == null) {
+ logger.warn("The RecoverableXAResource service with id {} does not have a name and will be ignored",
+ reference.getProperty("service.id"));
+ return null;
+ }
+
+ if(log == null) {
+ logger.warn("A RecoverableXAResource with id {} has been registered, but recovery logging is disabled for this Transaction Control service. No recovery will be availble in the event of a Transaction Manager failure.", resource.getId());
+ }
+
+ transactionManager.registerNamedXAResourceFactory(new NamedXAResourceFactory() {
+
+ @Override
+ public void returnNamedXAResource(NamedXAResource namedXAResource) {
+ resource.releaseXAResource(((NamedXAResourceImpl)namedXAResource).xaResource);
+ }
+
+ @Override
+ public NamedXAResource getNamedXAResource() throws SystemException {
+ try {
+ XAResource xaResource = resource.getXAResource();
+ if(xaResource == null) {
+ throw new IllegalStateException("The recoverable resource " + resource.getId()
+ + " is currently unavailable");
+ }
+ return new NamedXAResourceImpl(resource.getId(), xaResource,
+ transactionManager, false);
+ } catch (Exception e) {
+ throw new SystemException("Unable to get recoverable resource " +
+ resource.getId() + ": " + e.getMessage());
+ }
+ }
+
+ @Override
+ public String getName() {
+ return resource.getId();
+ }
+ });
+
+ return resource;
+ }
+
+ @Override
+ public void removedService(ServiceReference<RecoverableXAResource> reference,
+ RecoverableXAResource service) {
+ transactionManager.unregisterNamedXAResourceFactory(service.getId());
+ }
+
+ };
+ recoverableResources.open();
+ } else {
+ recoverableResources = null;
+ }
} catch (Exception e) {
destroy();
throw e;
@@ -73,7 +151,7 @@
}
private HOWLLog getLog(BundleContext ctx) throws Exception {
- Object recovery = config.getOrDefault("recovery.enabled", false);
+ Object recovery = config.getOrDefault("recovery.log.enabled", false);
if (recovery instanceof Boolean ? (Boolean) recovery : Boolean.valueOf(recovery.toString())) {
String logFileExt = "log";
@@ -122,6 +200,9 @@
}
public void destroy() {
+ if(recoverableResources != null) {
+ recoverableResources.close();
+ }
if(log != null) {
try {
log.doStop();
@@ -172,7 +253,7 @@
Map<String, Object> filtered = new HashMap<>();
copy(raw, filtered, "transaction.timeout");
- copy(raw, filtered, "recovery.enabled");
+ copy(raw, filtered, "recovery.log.enabled");
copy(raw, filtered, "recovery.log.dir");
copy(raw, filtered, "local.resources");
diff --git a/tx-control/tx-control-service-xa/src/main/java/org/apache/geronimo/transaction/manager/RecoveryWorkAroundTransactionManager.java b/tx-control/tx-control-service-xa/src/main/java/org/apache/geronimo/transaction/manager/RecoveryWorkAroundTransactionManager.java
new file mode 100644
index 0000000..d4c380f
--- /dev/null
+++ b/tx-control/tx-control-service-xa/src/main/java/org/apache/geronimo/transaction/manager/RecoveryWorkAroundTransactionManager.java
@@ -0,0 +1,17 @@
+package org.apache.geronimo.transaction.manager;
+
+import javax.transaction.xa.XAException;
+
+import org.apache.geronimo.transaction.log.HOWLLog;
+
+public class RecoveryWorkAroundTransactionManager extends GeronimoTransactionManager {
+
+ public RecoveryWorkAroundTransactionManager(int timeout, XidFactory xidFactory,
+ HOWLLog log) throws XAException {
+ super(timeout, xidFactory, log);
+ }
+
+ public NamedXAResourceFactory getNamedResource(String name) {
+ return super.getNamedXAResourceFactory(name);
+ }
+}
diff --git a/tx-control/tx-control-service-xa/src/test/java/org/apache/aries/tx/control/service/xa/impl/TransactionContextTest.java b/tx-control/tx-control-service-xa/src/test/java/org/apache/aries/tx/control/service/xa/impl/TransactionContextTest.java
index 3b4e2a6..d782b94 100644
--- a/tx-control/tx-control-service-xa/src/test/java/org/apache/aries/tx/control/service/xa/impl/TransactionContextTest.java
+++ b/tx-control/tx-control-service-xa/src/test/java/org/apache/aries/tx/control/service/xa/impl/TransactionContextTest.java
@@ -42,7 +42,8 @@
import javax.transaction.xa.Xid;
import org.apache.aries.tx.control.service.common.impl.AbstractTransactionContextImpl;
-import org.apache.geronimo.transaction.manager.GeronimoTransactionManager;
+import org.apache.geronimo.transaction.manager.RecoveryWorkAroundTransactionManager;
+import org.apache.geronimo.transaction.manager.XidFactoryImpl;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -66,7 +67,11 @@
@Before
public void setUp() throws XAException {
- ctx = new TransactionContextImpl(new GeronimoTransactionManager(), false, ENFORCE_SINGLE);
+ ctx = new TransactionContextImpl(getTxMgr(), false, ENFORCE_SINGLE);
+ }
+
+ private RecoveryWorkAroundTransactionManager getTxMgr() throws XAException {
+ return new RecoveryWorkAroundTransactionManager(30, new XidFactoryImpl(), null);
}
@Test
@@ -87,7 +92,7 @@
@Test
public void testisReadOnlyTrue() throws XAException {
- ctx = new TransactionContextImpl(new GeronimoTransactionManager(), true, ENFORCE_SINGLE);
+ ctx = new TransactionContextImpl(getTxMgr(), true, ENFORCE_SINGLE);
assertTrue(ctx.isReadOnly());
}
@@ -113,13 +118,13 @@
@Test
public void testLocalResourceSupportEnabled() throws XAException {
- ctx = new TransactionContextImpl(new GeronimoTransactionManager(), true, ENABLED);
+ ctx = new TransactionContextImpl(getTxMgr(), true, ENABLED);
assertTrue(ctx.supportsLocal());
}
@Test
public void testLocalResourceSupportDisabled() throws XAException {
- ctx = new TransactionContextImpl(new GeronimoTransactionManager(), true, DISABLED);
+ ctx = new TransactionContextImpl(getTxMgr(), true, DISABLED);
assertFalse(ctx.supportsLocal());
}
@@ -130,7 +135,12 @@
@Test
public void testXAResourceRegistration() {
- ctx.registerXAResource(xaResource);
+ ctx.registerXAResource(xaResource, null);
+ }
+
+ @Test
+ public void testRecoverableXAResourceRegistration() {
+ ctx.registerXAResource(xaResource, "anId");
}
@Test
@@ -351,7 +361,7 @@
@Test
public void testNoLocalResourceCanBeAddedWhenDisabled() throws Exception {
- ctx = new TransactionContextImpl(new GeronimoTransactionManager(), true, DISABLED);
+ ctx = new TransactionContextImpl(getTxMgr(), true, DISABLED);
try {
ctx.registerLocalResource(localResource);
@@ -363,7 +373,7 @@
@Test
public void testMultipleLocalResourcesFirstFailsSoRollback() throws Exception {
- ctx = new TransactionContextImpl(new GeronimoTransactionManager(), true, ENABLED);
+ ctx = new TransactionContextImpl(getTxMgr(), true, ENABLED);
ctx.registerLocalResource(localResource);
@@ -388,7 +398,7 @@
@Test
public void testSingleXAResource() throws Exception {
- ctx.registerXAResource(xaResource);
+ ctx.registerXAResource(xaResource, null);
Mockito.doAnswer(i -> {
assertEquals(COMMITTING, ctx.getTransactionStatus());
@@ -411,7 +421,7 @@
@Test
public void testXAResourceRollbackOnly() throws Exception {
- ctx.registerXAResource(xaResource);
+ ctx.registerXAResource(xaResource, null);
ctx.setRollbackOnly();
Mockito.doAnswer(i -> {
@@ -435,7 +445,7 @@
@Test
public void testXAResourcePreCommitException() throws Exception {
- ctx.registerXAResource(xaResource);
+ ctx.registerXAResource(xaResource, null);
Mockito.doAnswer(i -> {
assertEquals(ROLLING_BACK, ctx.getTransactionStatus());
@@ -460,7 +470,7 @@
@Test
public void testXAResourcePostCommitException() throws Exception {
- ctx.registerXAResource(xaResource);
+ ctx.registerXAResource(xaResource, null);
Mockito.doAnswer(i -> {
assertEquals(COMMITTING, ctx.getTransactionStatus());
@@ -490,7 +500,7 @@
public void testLastParticipantSuccessSoCommit() throws Exception {
ctx.registerLocalResource(localResource);
- ctx.registerXAResource(xaResource);
+ ctx.registerXAResource(xaResource, null);
Mockito.doAnswer(i -> {
assertEquals(COMMITTING, ctx.getTransactionStatus());
@@ -522,7 +532,7 @@
public void testLastParticipantFailsSoRollback() throws Exception {
ctx.registerLocalResource(localResource);
- ctx.registerXAResource(xaResource);
+ ctx.registerXAResource(xaResource, null);
Mockito.doAnswer(i -> {
assertEquals(COMMITTING, ctx.getTransactionStatus());
diff --git a/tx-control/tx-control-service-xa/src/test/java/org/apache/aries/tx/control/service/xa/impl/TransactionLogTest.java b/tx-control/tx-control-service-xa/src/test/java/org/apache/aries/tx/control/service/xa/impl/TransactionLogTest.java
index 84a3e29..b1487f8 100644
--- a/tx-control/tx-control-service-xa/src/test/java/org/apache/aries/tx/control/service/xa/impl/TransactionLogTest.java
+++ b/tx-control/tx-control-service-xa/src/test/java/org/apache/aries/tx/control/service/xa/impl/TransactionLogTest.java
@@ -18,42 +18,75 @@
*/
package org.apache.aries.tx.control.service.xa.impl;
+import static javax.transaction.xa.XAResource.XA_OK;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.osgi.service.transaction.control.TransactionStatus.ROLLED_BACK;
+import java.io.File;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiConsumer;
import javax.sql.XAConnection;
+import javax.transaction.xa.XAResource;
import org.h2.jdbcx.JdbcDataSource;
+import org.h2.tools.Server;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.Mockito;
import org.mockito.runners.MockitoJUnitRunner;
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.ServiceEvent;
+import org.osgi.framework.ServiceListener;
+import org.osgi.framework.ServiceReference;
+import org.osgi.service.transaction.control.TransactionException;
+import org.osgi.service.transaction.control.TransactionRolledBackException;
+import org.osgi.service.transaction.control.TransactionStatus;
+import org.osgi.service.transaction.control.recovery.RecoverableXAResource;
+/**
+ * The tests in this class look a little odd because we're using an
+ * unmanaged resource. This is to avoid creating a dependency on a
+ * JDBCResourceProvider just for the tests, and to give explicit
+ * control of when things get registered
+ *
+ */
@RunWith(MockitoJUnitRunner.class)
public class TransactionLogTest {
+ @Mock
+ BundleContext ctx;
+
+ @Mock
+ ServiceReference<RecoverableXAResource> serviceRef;
+
TransactionControlImpl txControl;
JdbcDataSource dataSource;
-
+
+ Server server;
+
@Before
public void setUp() throws Exception {
Map<String, Object> config = new HashMap<>();
- config.put("recovery.enabled", true);
- config.put("recovery.log.dir", "target/generated/recoverylog");
+ config.put("recovery.log.enabled", true);
+ config.put("recovery.log.dir", "target/recovery-test/recoverylog");
- txControl = new TransactionControlImpl(null, config);
+ txControl = new TransactionControlImpl(ctx, config);
- dataSource = new JdbcDataSource();
- dataSource.setUrl("jdbc:h2:mem:test;DB_CLOSE_DELAY=-1");
+ setupServerAndDataSource();
try (Connection conn = dataSource.getConnection()) {
Statement s = conn.createStatement();
@@ -61,21 +94,49 @@
s.execute("CREATE TABLE TEST_TABLE ( message varchar(255) )");
}
}
+
+ private void setupServerAndDataSource() throws SQLException {
+ server = Server.createTcpServer("-tcpPort", "0");
+ server.start();
+
+ File dbPath = new File("target/recovery-test/database");
+
+ dataSource = new JdbcDataSource();
+ dataSource.setUrl("jdbc:h2:tcp://127.0.0.1:" + server.getPort() + "/" + dbPath.getAbsolutePath());
+ }
@After
public void destroy() {
txControl.destroy();
+ try (Connection conn = dataSource.getConnection()) {
+ conn.createStatement().execute("shutdown immediately");
+ } catch (SQLException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+
+ delete(new File("target/recovery-test"));
+ }
+
+ private void delete(File file) {
+ if(file.isDirectory()) {
+ for(File f : file.listFiles()) {
+ delete(f);
+ }
+ }
+ file.delete();
}
@Test
- public void testRequired() throws Exception {
+ public void testRequiredNoRecovery() throws Exception {
XAConnection xaConn = dataSource.getXAConnection();
try {
txControl.required(() -> {
- txControl.getCurrentContext().registerXAResource(xaConn.getXAResource());
+ txControl.getCurrentContext().registerXAResource(xaConn.getXAResource(), null);
Connection conn = xaConn.getConnection();
+ // conn.setAutoCommit(false);
return conn.createStatement()
.execute("Insert into TEST_TABLE values ( 'Hello World!' )");
@@ -89,18 +150,216 @@
.executeQuery("Select * from TEST_TABLE");
rs.next();
assertEquals("Hello World!", rs.getString(1));
+ assertFalse(rs.next());
}
}
@Test
+ public void testRequired2PCNoRecovery() throws Exception {
+ XAConnection xaConn = dataSource.getXAConnection();
+ XAConnection xaConn2 = dataSource.getXAConnection();
+ try {
+ txControl.required(() -> {
+
+ txControl.getCurrentContext().registerXAResource(xaConn.getXAResource(), null);
+ txControl.getCurrentContext().registerXAResource(xaConn2.getXAResource(), null);
+
+ Connection conn = xaConn.getConnection();
+ // conn.setAutoCommit(false);
+ Connection conn2 = xaConn2.getConnection();
+ conn2.setAutoCommit(false);
+
+ conn.createStatement()
+ .execute("Insert into TEST_TABLE values ( 'Hello World!' )");
+ return conn2.createStatement()
+ .execute("Insert into TEST_TABLE values ( 'Hello World 2!' )");
+ });
+ } finally {
+ xaConn.close();
+ }
+
+ try (Connection conn = dataSource.getConnection()) {
+ ResultSet rs = conn.createStatement()
+ .executeQuery("Select * from TEST_TABLE order by message DESC");
+ rs.next();
+ assertEquals("Hello World!", rs.getString(1));
+ rs.next();
+ assertEquals("Hello World 2!", rs.getString(1));
+ assertFalse(rs.next());
+ }
+ }
+
+ @Test
+ public void testRequiredRecoverable() throws Exception {
+ XAConnection xaConn = dataSource.getXAConnection();
+ try {
+ txControl.required(() -> {
+
+ txControl.getCurrentContext().registerXAResource(xaConn.getXAResource(), "foo");
+
+ Connection conn = xaConn.getConnection();
+ // conn.setAutoCommit(false);
+
+ return conn.createStatement()
+ .execute("Insert into TEST_TABLE values ( 'Hello World!' )");
+ });
+ } finally {
+ xaConn.close();
+ }
+
+ try (Connection conn = dataSource.getConnection()) {
+ ResultSet rs = conn.createStatement()
+ .executeQuery("Select * from TEST_TABLE");
+ rs.next();
+ assertEquals("Hello World!", rs.getString(1));
+ }
+ }
+
+ @Test
+ public void testRequiredRecoveryRequiredPrePrepare() throws Exception {
+ doRecoveryRequired((good, poison) -> {
+ txControl.getCurrentContext().registerXAResource(poison, null);
+ txControl.getCurrentContext().registerXAResource(good, "foo");
+ }, TransactionStatus.ROLLED_BACK);
+
+ boolean success = false;
+ XAConnection conn = dataSource.getXAConnection();
+ for(int i=0; i < 5; i++) {
+ if(conn.getXAResource().recover(XAResource.TMSTARTRSCAN).length == 0) {
+ success = true;
+ break;
+ } else {
+ // Wait for recovery to happen!
+ Thread.sleep(500);
+ }
+ }
+
+ assertTrue("No recovery in time", success);
+ }
+
+ @Test
+ public void testRequiredRecoveryRequiredPostPrepare() throws Exception {
+ doRecoveryRequired((good, poison) -> {
+ txControl.getCurrentContext().registerXAResource(good, "foo");
+ txControl.getCurrentContext().registerXAResource(poison, null);
+ }, TransactionStatus.COMMITTED);
+
+ boolean success = false;
+ for(int i=0; i < 5; i++) {
+ try (Connection conn = dataSource.getConnection()) {
+ ResultSet rs = conn.createStatement()
+ .executeQuery("Select * from TEST_TABLE");
+ if(rs.next()) {
+ assertEquals("Hello World!", rs.getString(1));
+ success = true;
+ break;
+ } else {
+ // Wait for recovery to happen!
+ Thread.sleep(500);
+ }
+ }
+ }
+
+ assertTrue("No recovery in time", success);
+ }
+
+ public void doRecoveryRequired(BiConsumer<XAResource, XAResource> ordering,
+ TransactionStatus expectedFinalState) throws Exception {
+
+ //Register the recoverable resource
+ ArgumentCaptor<ServiceListener> captor = ArgumentCaptor.forClass(ServiceListener.class);
+ Mockito.verify(ctx).addServiceListener(captor.capture(), Mockito.anyString());
+ Mockito.when(ctx.getService(serviceRef)).thenReturn(new TestRecoverableResource("foo", dataSource));
+
+ captor.getValue().serviceChanged(new ServiceEvent(ServiceEvent.REGISTERED, serviceRef));
+
+ XAConnection xaConn = dataSource.getXAConnection();
+ AtomicReference<TransactionStatus> ref = new AtomicReference<TransactionStatus>();
+ try {
+ txControl.required(() -> {
+
+ txControl.getCurrentContext().postCompletion(ref::set);
+
+ Connection conn = xaConn.getConnection();
+ // conn.setAutoCommit(false);
+
+ XAResource dsResource = xaConn.getXAResource();
+
+ XAResource poison = Mockito.mock(XAResource.class);
+ Mockito.when(poison.prepare(Mockito.any())).thenAnswer(i -> {
+ // Now kill the db server before it commits!
+ conn.createStatement().execute("shutdown immediately");
+ Thread.sleep(1000);
+ return XA_OK;
+ });
+
+ ordering.accept(dsResource, poison);
+
+ return conn.createStatement()
+ .execute("Insert into TEST_TABLE values ( 'Hello World!' )");
+ });
+ } catch (TransactionException te) {
+ assertEquals(expectedFinalState, ref.get());
+ assertEquals(expectedFinalState == ROLLED_BACK, te instanceof TransactionRolledBackException);
+ } finally {
+ try {
+ xaConn.close();
+ } catch (SQLException sqle) {}
+ }
+
+ setupServerAndDataSource();
+
+ }
+
+ static class TestRecoverableResource implements RecoverableXAResource {
+
+ private final String id;
+
+ private final JdbcDataSource dataSource;
+
+ public TestRecoverableResource(String id, JdbcDataSource dataSource) {
+ this.id = id;
+ this.dataSource = dataSource;
+ }
+
+ @Override
+ public String getId() {
+ return id;
+ }
+
+ @Override
+ public XAResource getXAResource() throws Exception {
+ XAConnection xaConnection = dataSource.getXAConnection();
+ if(xaConnection.getConnection().isValid(2)) {
+ return xaConnection.getXAResource();
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public void releaseXAResource(XAResource xaRes) {
+ // This is valid for H2;
+ try {
+ ((XAConnection) xaRes).close();
+ } catch (SQLException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ }
+
+ }
+
+ @Test
public void testRequiredWithRollback() throws Exception {
XAConnection xaConn = dataSource.getXAConnection();
try {
txControl.required(() -> {
- txControl.getCurrentContext().registerXAResource(xaConn.getXAResource());
+ txControl.getCurrentContext().registerXAResource(xaConn.getXAResource(), null);
Connection conn = xaConn.getConnection();
+ // conn.setAutoCommit(false);
conn.createStatement()
.execute("Insert into TEST_TABLE values ( 'Hello World!' )");
