[jboss-svn-commits] JBL Code SVN: r37564 - in labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed: server and 1 other directories.
jboss-svn-commits at lists.jboss.org
jboss-svn-commits at lists.jboss.org
Sun Oct 16 17:31:39 EDT 2011
Author: tomjenkinson
Date: 2011-10-16 17:31:39 -0400 (Sun, 16 Oct 2011)
New Revision: 37564
Modified:
labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/SimpleIsolatedServers.java
labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/TestResource.java
labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/IsolatableServersClassLoader.java
labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/LocalServer.java
labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/RemoteServer.java
labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/impl/ProxyXAResource.java
labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/impl/ServerImpl.java
Log:
JBTM-916 updated to separate all usages of a server by its classloader
Modified: labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/SimpleIsolatedServers.java
===================================================================
--- labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/SimpleIsolatedServers.java 2011-10-16 21:20:56 UTC (rev 37563)
+++ labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/SimpleIsolatedServers.java 2011-10-16 21:31:39 UTC (rev 37564)
@@ -35,19 +35,21 @@
@BeforeClass
public static void setup() throws SecurityException, NoSuchMethodException, InstantiationException, IllegalAccessException, ClassNotFoundException,
CoreEnvironmentBeanException, IOException, IllegalArgumentException, NoSuchFieldException {
- ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
for (int i = 0; i < localServers.length; i++) {
+ ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
IsolatableServersClassLoader classLoader = new IsolatableServersClassLoader("com.arjuna.ats.jta.distributed.server", contextClassLoader);
localServers[i] = (LocalServer) classLoader.loadClass("com.arjuna.ats.jta.distributed.server.impl.ServerImpl").newInstance();
+ Thread.currentThread().setContextClassLoader(localServers[i].getClass().getClassLoader());
localServers[i].initialise(lookupProvider, (i + 1) * 1000);
remoteServers[i] = localServers[i].connectTo();
+ Thread.currentThread().setContextClassLoader(contextClassLoader);
}
}
@Test
public void testRecovery() {
- getLocalServer(3000).doRecoveryManagerScan();
- getLocalServer(2000).doRecoveryManagerScan();
+ // getLocalServer(3000).doRecoveryManagerScan();
+ // getLocalServer(2000).doRecoveryManagerScan();
getLocalServer(1000).doRecoveryManagerScan();
}
@@ -59,86 +61,37 @@
if (file.exists()) {
file.delete();
}
- int startingTimeout = 10;
+ int startingTimeout = 10000;
+ List<Integer> nodesToFlowTo = new LinkedList<Integer>(Arrays.asList(new Integer[] { 1000, 2000, 3000, 2000, 1000, 2000, 3000, 1000, 3000 }));
// Start out at the first server
LocalServer originalServer = getLocalServer(1000);
+ ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+ Thread.currentThread().setContextClassLoader(originalServer.getClass().getClassLoader());
TransactionManager transactionManager = originalServer.getTransactionManager();
transactionManager.setTransactionTimeout(startingTimeout);
transactionManager.begin();
Transaction originalTransaction = transactionManager.getTransaction();
- originalTransaction.registerSynchronization(new TestSynchronization(originalServer.getNodeName()));
- originalTransaction.enlistResource(new TestResource(originalServer.getNodeName(), false));
- Xid toMigrate = originalServer.storeCurrentTransaction();
-
- // Loop through the rest of the servers passing the transaction up and
- // down
- Transaction suspendedTransaction = transactionManager.suspend();
- long timeLeftBeforeTransactionTimeout = originalServer.getTimeLeftBeforeTransactionTimeout();
- List<Integer> nodesToFlowTo = new LinkedList<Integer>(Arrays.asList(new Integer[] { 2000, 3000, 2000, 1000, 2000, 3000, 1000, 3000 }));
- boolean proxyRequired = recursivelyFlowTransaction(nodesToFlowTo, timeLeftBeforeTransactionTimeout, toMigrate);
- transactionManager.resume(suspendedTransaction);
- if (proxyRequired) {
- XAResource proxyXAResource = originalServer.generateProxyXAResource(lookupProvider, originalServer.getNodeName(), 2000);
- originalTransaction.enlistResource(proxyXAResource);
- originalTransaction.registerSynchronization(originalServer.generateProxySynchronization(lookupProvider, originalServer.getNodeName(), 2000,
- toMigrate));
- }
-
- Transaction transaction = transactionManager.getTransaction();
- transaction.commit();
- originalServer.removeTransaction(toMigrate);
+ int remainingTimeout = (int) (originalServer.getTimeLeftBeforeTransactionTimeout() / 1000);
+ Xid currentXid = originalServer.getCurrentXid();
+ originalServer.storeRootTransaction();
+ transactionManager.suspend();
+ performTransactionalWork(nodesToFlowTo, remainingTimeout, currentXid);
+ transactionManager.resume(originalTransaction);
+ originalServer.removeRootTransaction(currentXid);
+ originalTransaction.commit();
+ Thread.currentThread().setContextClassLoader(classLoader);
}
- @Test
- public void testMigrateTransactionRollback() throws NotSupportedException, SystemException, IllegalStateException, RollbackException,
- InvalidTransactionException, XAException, SecurityException, HeuristicMixedException, HeuristicRollbackException {
-
- File file = new File(System.getProperty("user.dir") + "/tmp/");
- if (file.exists()) {
- file.delete();
- }
- int startingTimeout = 10;
-
- // Start out at the first server
- LocalServer originalServer = getLocalServer(1000);
- TransactionManager transactionManager = originalServer.getTransactionManager();
- transactionManager.setTransactionTimeout(startingTimeout);
- transactionManager.begin();
- Transaction originalTransaction = transactionManager.getTransaction();
- originalTransaction.registerSynchronization(new TestSynchronization(originalServer.getNodeName()));
- originalTransaction.enlistResource(new TestResource(originalServer.getNodeName(), false));
- Xid toMigrate = originalServer.storeCurrentTransaction();
-
- // Loop through the rest of the servers passing the transaction up and
- // down
- Transaction suspendedTransaction = transactionManager.suspend();
- long timeLeftBeforeTransactionTimeout = originalServer.getTimeLeftBeforeTransactionTimeout();
- List<Integer> nodesToFlowTo = new LinkedList<Integer>(Arrays.asList(new Integer[] { 2000, 3000, 2000, 1000, 2000, 3000, 1000, 3000 }));
- boolean proxyRequired = recursivelyFlowTransaction(nodesToFlowTo, timeLeftBeforeTransactionTimeout, toMigrate);
- transactionManager.resume(suspendedTransaction);
- if (proxyRequired) {
- XAResource proxyXAResource = originalServer.generateProxyXAResource(lookupProvider, originalServer.getNodeName(), 2000);
- originalTransaction.enlistResource(proxyXAResource);
- originalTransaction.registerSynchronization(originalServer.generateProxySynchronization(lookupProvider, originalServer.getNodeName(), 2000,
- toMigrate));
- }
-
- Transaction transaction = transactionManager.getTransaction();
- transaction.rollback();
- originalServer.removeTransaction(toMigrate);
- }
-
- private boolean recursivelyFlowTransaction(List<Integer> nodesToFlowTo, long timeLeftBeforeTransactionTimeout, Xid toMigrate) throws RollbackException,
+ private boolean performTransactionalWork(List<Integer> nodesToFlowTo, int remainingTimeout, Xid toMigrate) throws RollbackException,
InvalidTransactionException, IllegalStateException, XAException, SystemException, NotSupportedException {
-
Integer currentServerName = nodesToFlowTo.remove(0);
LocalServer currentServer = getLocalServer(currentServerName);
- // Migrate the transaction to the next server
- int remainingTimeout = (int) (timeLeftBeforeTransactionTimeout / 1000);
+ ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+ Thread.currentThread().setContextClassLoader(currentServer.getClass().getClassLoader());
- boolean requiresProxyAtPreviousServer = !currentServer.getTransaction(remainingTimeout, toMigrate);
+ boolean requiresProxyAtPreviousServer = !currentServer.getAndResumeTransaction(remainingTimeout, toMigrate);
// Perform work on the migrated transaction
TransactionManager transactionManager = currentServer.getTransactionManager();
Transaction transaction = transactionManager.getTransaction();
@@ -148,32 +101,33 @@
if (!nodesToFlowTo.isEmpty()) {
Integer nextServerNodeName = nodesToFlowTo.get(0);
- // SUSPEND THE TRANSACTION
- Transaction suspendedTransaction = transactionManager.suspend();
// FLOW THE TRANSACTION
- timeLeftBeforeTransactionTimeout = currentServer.getTimeLeftBeforeTransactionTimeout();
- boolean proxyRequired = recursivelyFlowTransaction(nodesToFlowTo, timeLeftBeforeTransactionTimeout, toMigrate);
+ remainingTimeout = (int) (currentServer.getTimeLeftBeforeTransactionTimeout() / 1000);
+
+ // SUSPEND THE TRANSACTION
+ Xid currentXid = currentServer.getCurrentXid();
+ transactionManager.suspend();
+ boolean proxyRequired = performTransactionalWork(nodesToFlowTo, remainingTimeout, currentXid);
+ transactionManager.resume(transaction);
+
// Create a proxy for the new server if necessary, this can orphan
- // the
- // remote server but XA recovery will handle that on the remote
+ // the remote server but XA recovery will handle that on the remote
// server
// The alternative is to always create a proxy but this is a
- // performance
- // drain and will result in multiple subordinate transactions and
- // performance
- // issues
- // RESUME THE TRANSACTION IN CASE THERE IS MORE WORK
- transactionManager.resume(suspendedTransaction);
+ // performance drain and will result in multiple subordinate
+ // transactions and performance issues
if (proxyRequired) {
XAResource proxyXAResource = currentServer.generateProxyXAResource(lookupProvider, currentServer.getNodeName(), nextServerNodeName);
- suspendedTransaction.enlistResource(proxyXAResource);
- suspendedTransaction.registerSynchronization(currentServer.generateProxySynchronization(lookupProvider, currentServer.getNodeName(),
- nextServerNodeName, toMigrate));
+ transaction.enlistResource(proxyXAResource);
+ transaction.registerSynchronization(currentServer.generateProxySynchronization(lookupProvider, currentServer.getNodeName(), nextServerNodeName,
+ toMigrate));
}
}
// SUSPEND THE TRANSACTION WHEN YOU ARE READY TO RETURN TO YOUR CALLER
transactionManager.suspend();
+
+ Thread.currentThread().setContextClassLoader(classLoader);
return requiresProxyAtPreviousServer;
}
Modified: labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/TestResource.java
===================================================================
--- labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/TestResource.java 2011-10-16 21:20:56 UTC (rev 37563)
+++ labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/TestResource.java 2011-10-16 21:31:39 UTC (rev 37564)
@@ -124,6 +124,7 @@
if (file != null) {
file.delete();
}
+ this.xid = null;
}
public synchronized void rollback(Xid xid) throws XAException {
@@ -131,6 +132,7 @@
if (file != null) {
file.delete();
}
+ this.xid = null;
}
public void end(Xid xid, int flags) throws XAException {
@@ -171,7 +173,11 @@
if (flag == XAResource.TMNOFLAGS) {
System.out.println(" TestResource (" + serverId + ") RECOVER[XAResource.TMENDRSCAN]: " + serverId);
}
- return new Xid[] { xid };
+ if (xid == null) {
+ return null;
+ } else {
+ return new Xid[] { xid };
+ }
}
public boolean setTransactionTimeout(int seconds) throws XAException {
Modified: labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/IsolatableServersClassLoader.java
===================================================================
--- labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/IsolatableServersClassLoader.java 2011-10-16 21:20:56 UTC (rev 37563)
+++ labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/IsolatableServersClassLoader.java 2011-10-16 21:31:39 UTC (rev 37564)
@@ -46,7 +46,6 @@
if (clazzMap.containsKey(name)) {
clazz = clazzMap.get(name);
}
-
if (!name.startsWith("com.arjuna") || (ignoredPackage != null && name.matches(ignoredPackage + ".[A-Za-z0-9]*"))) {
clazz = super.loadClass(name);
} else {
Modified: labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/LocalServer.java
===================================================================
--- labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/LocalServer.java 2011-10-16 21:20:56 UTC (rev 37563)
+++ labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/LocalServer.java 2011-10-16 21:31:39 UTC (rev 37564)
@@ -27,11 +27,11 @@
public long getTimeLeftBeforeTransactionTimeout() throws RollbackException;
- public Xid storeCurrentTransaction() throws SystemException;
+ public void storeRootTransaction() throws SystemException;
- public void removeTransaction(Xid toMigrate);
+ public void removeRootTransaction(Xid toMigrate);
- public boolean getTransaction(int remainingTimeout, Xid toImport) throws XAException, InvalidTransactionException, IllegalStateException, SystemException;
+ public boolean getAndResumeTransaction(int remainingTimeout, Xid toImport) throws XAException, InvalidTransactionException, IllegalStateException, SystemException;
public RemoteServer connectTo();
@@ -40,4 +40,8 @@
public XAResource generateProxyXAResource(LookupProvider lookupProvider, Integer localServerName, Integer remoteServerName);
public Synchronization generateProxySynchronization(LookupProvider lookupProvider, Integer localServerName, Integer remoteServerName, Xid toRegisterAgainst);
+
+ public Xid extractXid(XAResource proxyXAResource);
+
+ public Xid getCurrentXid() throws SystemException;
}
Modified: labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/RemoteServer.java
===================================================================
--- labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/RemoteServer.java 2011-10-16 21:20:56 UTC (rev 37563)
+++ labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/RemoteServer.java 2011-10-16 21:31:39 UTC (rev 37564)
@@ -1,7 +1,5 @@
package com.arjuna.ats.jta.distributed.server;
-import java.util.List;
-
import javax.transaction.HeuristicCommitException;
import javax.transaction.HeuristicMixedException;
import javax.transaction.HeuristicRollbackException;
@@ -19,7 +17,7 @@
public void propagateRollback(Xid xid) throws IllegalStateException, HeuristicMixedException, HeuristicCommitException, HeuristicRollbackException,
SystemException, XAException, DummyRemoteException;
- public Xid[] propagateRecover(List<Integer> startScanned, int flag) throws XAException, DummyRemoteException;
+ public Xid[] propagateRecover(Integer serverNodeNameToRecoverFor, int flag) throws XAException, DummyRemoteException;
public void propagateForget(Xid xid) throws XAException, DummyRemoteException;
Modified: labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/impl/ProxyXAResource.java
===================================================================
--- labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/impl/ProxyXAResource.java 2011-10-16 21:20:56 UTC (rev 37563)
+++ labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/impl/ProxyXAResource.java 2011-10-16 21:31:39 UTC (rev 37564)
@@ -6,8 +6,6 @@
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
import javax.transaction.HeuristicCommitException;
import javax.transaction.HeuristicMixedException;
@@ -25,15 +23,12 @@
public class ProxyXAResource implements XAResource, XAResourceWrapper {
- public static final ThreadLocal<List<Integer>> RECOVERY_SCAN_STARTED = new ThreadLocal<List<Integer>>();
-
private int transactionTimeout;
- private Xid xid;
private Integer remoteServerName = -1;
private File file;
private Integer localServerName;
-
private LookupProvider lookupProvider;
+ private Xid xid;
public ProxyXAResource(LookupProvider lookupProvider, Integer localServerName, Integer remoteServerName) {
this.lookupProvider = lookupProvider;
@@ -46,19 +41,18 @@
this.localServerName = localServerName;
this.file = file;
DataInputStream fis = new DataInputStream(new FileInputStream(file));
+ this.remoteServerName = fis.readInt();
final int formatId = fis.readInt();
- final int gtrid_length = fis.readInt();
+ int gtrid_length = fis.readInt();
final byte[] gtrid = new byte[gtrid_length];
fis.read(gtrid, 0, gtrid_length);
- final int bqual_length = fis.readInt();
+ int bqual_length = fis.readInt();
final byte[] bqual = new byte[bqual_length];
fis.read(bqual, 0, bqual_length);
- int remoteServerName = fis.readInt();
- this.remoteServerName = remoteServerName;
this.xid = new Xid() {
@Override
- public byte[] getGlobalTransactionId() {
- return gtrid;
+ public byte[] getBranchQualifier() {
+ return bqual;
}
@Override
@@ -67,52 +61,12 @@
}
@Override
- public byte[] getBranchQualifier() {
- return bqual;
+ public byte[] getGlobalTransactionId() {
+ return gtrid;
}
-
- @Override
- public boolean equals(Object object) {
- Xid xid = (Xid) object;
- if (xid == null)
- return false;
-
- if (xid == this)
- return true;
- else {
-
- if (xid.getFormatId() == formatId) {
- byte[] gtx = xid.getGlobalTransactionId();
- byte[] bql = xid.getBranchQualifier();
- final int bqlength = (bql == null ? 0 : bql.length);
-
- if ((gtrid.length == gtx.length) && (bqual.length == bqlength)) {
- int i;
-
- for (i = 0; i < gtrid.length; i++) {
- if (gtrid[i] != gtx[i])
- return false;
- }
-
- for (i = 0; i < bqual.length; i++) {
- if (bqual[i] != bql[i])
- return false;
- }
-
- return true;
- }
- }
- }
-
- return false;
- }
};
}
- public Xid getXid() {
- return xid;
- }
-
@Override
public void start(Xid xid, int flags) throws XAException {
System.out.println(" ProxyXAResource (" + localServerName + ":" + remoteServerName + ") XA_START [" + xid + "]");
@@ -133,41 +87,40 @@
File dir = new File(System.getProperty("user.dir") + "/tmp/ProxyXAResource/" + localServerName + "/");
dir.mkdirs();
file = new File(dir, new Uid().fileStringForm());
-
file.createNewFile();
-
- final int formatId = xid.getFormatId();
- final byte[] gtrid = xid.getGlobalTransactionId();
- final int gtrid_length = gtrid.length;
- final byte[] bqual = xid.getBranchQualifier();
- final int bqual_length = bqual.length;
-
DataOutputStream fos = new DataOutputStream(new FileOutputStream(file));
- fos.writeInt(formatId);
- fos.writeInt(gtrid_length);
- fos.write(gtrid, 0, gtrid_length);
- fos.writeInt(bqual_length);
- fos.write(bqual, 0, bqual_length);
fos.writeInt(remoteServerName);
+ fos.writeInt(xid.getFormatId());
+ fos.writeInt(xid.getGlobalTransactionId().length);
+ fos.write(xid.getGlobalTransactionId());
+ fos.writeInt(xid.getBranchQualifier().length);
+ fos.write(xid.getBranchQualifier());
} catch (IOException e) {
e.printStackTrace();
throw new XAException(XAException.XAER_RMERR);
}
+ ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
try {
+ Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
int propagatePrepare = lookupProvider.lookup(remoteServerName).propagatePrepare(xid);
System.out.println(" ProxyXAResource (" + localServerName + ":" + remoteServerName + ") XA_PREPARED");
return propagatePrepare;
} catch (DummyRemoteException ce) {
throw new XAException(XAException.XA_RETRY);
+ } finally {
+ Thread.currentThread().setContextClassLoader(contextClassLoader);
}
+
}
@Override
public synchronized void commit(Xid xid, boolean onePhase) throws XAException {
System.out.println(" ProxyXAResource (" + localServerName + ":" + remoteServerName + ") XA_COMMIT [" + xid + "]");
+ ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
try {
+ Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
lookupProvider.lookup(remoteServerName).propagateCommit(xid, onePhase);
System.out.println(" ProxyXAResource (" + localServerName + ":" + remoteServerName + ") XA_COMMITED");
} catch (IllegalStateException e) {
@@ -182,6 +135,8 @@
throw new XAException(XAException.XAER_PROTO);
} catch (DummyRemoteException ce) {
throw new XAException(XAException.XA_RETRY);
+ } finally {
+ Thread.currentThread().setContextClassLoader(contextClassLoader);
}
if (file != null) {
@@ -192,7 +147,9 @@
@Override
public synchronized void rollback(Xid xid) throws XAException {
System.out.println(" ProxyXAResource (" + localServerName + ":" + remoteServerName + ") XA_ROLLBACK[" + xid + "]");
+ ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
try {
+ Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
lookupProvider.lookup(remoteServerName).propagateRollback(xid);
System.out.println(" ProxyXAResource (" + localServerName + ":" + remoteServerName + ") XA_ROLLBACKED");
} catch (IllegalStateException e) {
@@ -207,6 +164,8 @@
throw new XAException(XAException.XAER_PROTO);
} catch (DummyRemoteException ce) {
throw new XAException(XAException.XA_RETRY);
+ } finally {
+ Thread.currentThread().setContextClassLoader(contextClassLoader);
}
if (file != null) {
@@ -216,64 +175,58 @@
@Override
public Xid[] recover(int flag) throws XAException {
- List<Integer> startScanned = RECOVERY_SCAN_STARTED.get();
- if (startScanned == null) {
- startScanned = new ArrayList<Integer>();
- RECOVERY_SCAN_STARTED.set(startScanned);
- }
-
- int tocheck = (flag & XAResource.TMSTARTRSCAN);
- if (tocheck == XAResource.TMSTARTRSCAN) {
+ Xid[] recovered = null;
+ if ((flag & XAResource.TMSTARTRSCAN) == XAResource.TMSTARTRSCAN) {
System.out.println(" ProxyXAResource (" + localServerName + ":" + remoteServerName + ") XA_RECOVER [XAResource.TMSTARTRSCAN]: "
+ remoteServerName);
+ }
+ if ((flag & XAResource.TMSTARTRSCAN) == XAResource.TMENDRSCAN) {
+ System.out.println(" ProxyXAResource (" + localServerName + ":" + remoteServerName + ") XA_RECOVER [XAResource.TMENDRSCAN]: "
+ + remoteServerName);
+ }
- if (!startScanned.contains(remoteServerName)) {
- startScanned.add(remoteServerName);
+ ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
+ try {
+ Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
+ recovered = lookupProvider.lookup(remoteServerName).propagateRecover(localServerName, flag);
+ } catch (DummyRemoteException ce) {
+ throw new XAException(XAException.XA_RETRY);
+ } finally {
+ Thread.currentThread().setContextClassLoader(contextClassLoader);
+ }
- // Make sure that the remote server has recovered all
- // transactions
- try {
- lookupProvider.lookup(remoteServerName).propagateRecover(startScanned, flag);
- } catch (DummyRemoteException ce) {
- throw new XAException(XAException.XA_RETRY);
- } finally {
- startScanned.remove((Integer) remoteServerName);
- }
- }
+ for (int i = 0; i < recovered.length; i++) {
+ System.out.println(" ProxyXAResource (" + localServerName + ":" + remoteServerName + ") recovered: " + recovered[i]);
+ }
+ if ((flag & XAResource.TMSTARTRSCAN) == XAResource.TMSTARTRSCAN) {
System.out.println(" ProxyXAResource (" + localServerName + ":" + remoteServerName + ") XA_RECOVERD[XAResource.TMSTARTRSCAN]: "
+ remoteServerName);
}
- tocheck = (flag & XAResource.TMENDRSCAN);
- if (tocheck == XAResource.TMENDRSCAN) {
- System.out.println(" ProxyXAResource (" + localServerName + ":" + remoteServerName + ") XA_RECOVER [XAResource.TMENDRSCAN]: "
- + remoteServerName);
-
- if (!startScanned.contains(remoteServerName)) {
- try {
- lookupProvider.lookup(remoteServerName).propagateRecover(startScanned, flag);
- } catch (DummyRemoteException ce) {
- throw new XAException(XAException.XA_RETRY);
- } finally {
- startScanned.remove((Integer) remoteServerName);
- }
- }
-
+ if ((flag & XAResource.TMSTARTRSCAN) == XAResource.TMENDRSCAN) {
System.out.println(" ProxyXAResource (" + localServerName + ":" + remoteServerName + ") XA_RECOVERD[XAResource.TMENDRSCAN]: "
+ remoteServerName);
}
- return new Xid[] { xid };
+ if (recovered.length > 0) {
+ return new Xid[] { xid };
+ } else {
+ return null;
+ }
}
@Override
public void forget(Xid xid) throws XAException {
System.out.println(" ProxyXAResource (" + localServerName + ":" + remoteServerName + ") XA_FORGET [" + xid + "]");
+ ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
try {
+ Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
lookupProvider.lookup(remoteServerName).propagateForget(xid);
System.out.println(" ProxyXAResource (" + localServerName + ":" + remoteServerName + ") XA_FORGETED[" + xid + "]");
} catch (DummyRemoteException ce) {
throw new XAException(XAException.XA_RETRY);
+ } finally {
+ Thread.currentThread().setContextClassLoader(contextClassLoader);
}
}
@@ -290,7 +243,13 @@
@Override
public boolean isSameRM(XAResource xares) throws XAException {
- return xares.equals(this);
+ boolean toReturn = false;
+ if (xares instanceof ProxyXAResource) {
+ if (((ProxyXAResource) xares).remoteServerName == remoteServerName) {
+ toReturn = true;
+ }
+ }
+ return toReturn;
}
/**
@@ -321,4 +280,8 @@
public String getJndiName() {
return "ProxyXAResource";
}
+
+ public Xid getXid() {
+ return xid;
+ }
}
Modified: labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/impl/ServerImpl.java
===================================================================
--- labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/impl/ServerImpl.java 2011-10-16 21:20:56 UTC (rev 37563)
+++ labs/jbosstm/branches/JBOSSTS_4_15_0_Final/atsintegration/tests/classes/com/arjuna/ats/jta/distributed/server/impl/ServerImpl.java 2011-10-16 21:31:39 UTC (rev 37564)
@@ -1,7 +1,6 @@
package com.arjuna.ats.jta.distributed.server.impl;
import java.io.IOException;
-import java.lang.reflect.Field;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.HashMap;
@@ -32,7 +31,6 @@
import com.arjuna.ats.arjuna.recovery.RecoveryManager;
import com.arjuna.ats.arjuna.tools.osb.mbean.ObjStoreBrowser;
import com.arjuna.ats.internal.jbossatx.jta.XAResourceRecordWrappingPluginImpl;
-import com.arjuna.ats.internal.jta.recovery.arjunacore.RecoveryXids;
import com.arjuna.ats.internal.jta.transaction.arjunacore.TransactionImple;
import com.arjuna.ats.internal.jta.transaction.arjunacore.jca.SubordinateTransaction;
import com.arjuna.ats.internal.jta.transaction.arjunacore.jca.SubordinateXidImple;
@@ -45,6 +43,8 @@
import com.arjuna.ats.jta.distributed.server.LocalServer;
import com.arjuna.ats.jta.distributed.server.LookupProvider;
import com.arjuna.ats.jta.distributed.server.RemoteServer;
+import com.arjuna.ats.jta.xa.XATxConverter;
+import com.arjuna.ats.jta.xa.XidImple;
public class ServerImpl implements LocalServer, RemoteServer {
@@ -140,11 +140,12 @@
.setTransactionSynchronizationRegistry(new com.arjuna.ats.internal.jta.transaction.arjunacore.TransactionSynchronizationRegistryImple());
transactionManagerService.create();
- Field safetyIntervalMillis = RecoveryXids.class.getDeclaredField("safetyIntervalMillis");
- safetyIntervalMillis.setAccessible(true);
- Field modifiersField = Field.class.getDeclaredField("modifiers");
- modifiersField.setAccessible(true);
- safetyIntervalMillis.set(null, 0);
+ // Field safetyIntervalMillis =
+ // RecoveryXids.class.getDeclaredField("safetyIntervalMillis");
+ // safetyIntervalMillis.setAccessible(true);
+ // Field modifiersField = Field.class.getDeclaredField("modifiers");
+ // modifiersField.setAccessible(true);
+ // safetyIntervalMillis.set(null, 0);
}
@Override
@@ -158,7 +159,7 @@
}
@Override
- public boolean getTransaction(int remainingTimeout, Xid toResume) throws XAException, InvalidTransactionException, IllegalStateException, SystemException {
+ public boolean getAndResumeTransaction(int remainingTimeout, Xid toResume) throws XAException, InvalidTransactionException, IllegalStateException, SystemException {
boolean existed = true;
Transaction transaction = transactions.get(new SubordinateXidImple(toResume));
if (transaction == null) {
@@ -183,20 +184,25 @@
}
@Override
- public Xid storeCurrentTransaction() throws SystemException {
+ public void storeRootTransaction() throws SystemException {
TransactionImple transaction = ((TransactionImple) transactionManagerService.getTransactionManager().getTransaction());
Xid txId = transaction.getTxId();
transactions.put(new SubordinateXidImple(txId), transaction);
- return txId;
}
@Override
- public void removeTransaction(Xid toMigrate) {
+ public Xid getCurrentXid() throws SystemException {
+ TransactionImple transaction = ((TransactionImple) transactionManagerService.getTransactionManager().getTransaction());
+ return transaction.getTxId();
+ }
+
+ @Override
+ public void removeRootTransaction(Xid toMigrate) {
transactions.remove(new SubordinateXidImple(toMigrate));
}
@Override
- public XAResource generateProxyXAResource(LookupProvider lookupProvider, Integer localServerName, Integer remoteServerName) {
+ public ProxyXAResource generateProxyXAResource(LookupProvider lookupProvider, Integer localServerName, Integer remoteServerName) {
return new ProxyXAResource(lookupProvider, localServerName, remoteServerName);
}
@@ -242,13 +248,22 @@
}
@Override
- public Xid[] propagateRecover(List<Integer> recoveryScanStarted, int flag) throws XAException, DummyRemoteException {
+ public Xid[] propagateRecover(Integer serverNodeNameToRecoverFor, int flag) throws XAException, DummyRemoteException {
if (offline) {
throw new DummyRemoteException("Connection refused to: " + nodeName);
}
// Assumes that this thread is used by the recovery thread
- ProxyXAResource.RECOVERY_SCAN_STARTED.set(recoveryScanStarted);
- return SubordinationManager.getXATerminator().recover(flag);
+ // ProxyXAResource.RECOVERY_SCAN_STARTED.set(recoveryScanStarted);
+ List<Xid> toReturn = new ArrayList<Xid>();
+ Xid[] recovered = SubordinationManager.getXATerminator().recover(flag);
+ if (recovered != null) {
+ for (int i = 0; i < recovered.length; i++) {
+ if (XATxConverter.getParentNodeName(((XidImple) recovered[i]).getXID()) == serverNodeNameToRecoverFor) {
+ toReturn.add(recovered[i]);
+ }
+ }
+ }
+ return toReturn.toArray(new Xid[0]);
}
@Override
@@ -268,4 +283,10 @@
SubordinateTransaction tx = SubordinationManager.getTransactionImporter().getImportedTransaction(xid);
tx.doBeforeCompletion();
}
+
+ @Override
+ public Xid extractXid(XAResource xaResource) {
+ ProxyXAResource proxyXAResource = (ProxyXAResource) xaResource;
+ return proxyXAResource.getXid();
+ }
}
More information about the jboss-svn-commits
mailing list