c1.put(k1, "v1");
c1.put(k2, "v2");
// We split the transaction commit in two phases by calling the TransactionCoordinator methods directly
TransactionTable txTable = TestingUtil.extractComponent(c1, TransactionTable.class);
TransactionCoordinator txCoordinator = TestingUtil.extractComponent(c1, TransactionCoordinator.class);
// Execute the prepare on both nodes
LocalTransaction localTx = txTable.getLocalTransaction(tm(c1).getTransaction());
txCoordinator.prepare(localTx);
// Delay the commit on the remote node. Can't used blockNewTransactions because we don't want a StateTransferInProgressException
InterceptorChain c2ic = TestingUtil.extractComponent(c2, InterceptorChain.class);
c2ic.addInterceptorBefore(new CommandInterceptor() {
protected Object handleDefault(InvocationContext ctx, VisitableCommand command) throws Throwable {
if (command instanceof CommitCommand) {
Thread.sleep(3000);
}
return super.handleDefault(ctx, command);
}
}, StateTransferInterceptor.class);
// Schedule the remote node to stop on another thread since the main thread will be busy with the commit call
Thread worker = new Thread("RehasherSim,StaleLocksWithCommitDuringStateTransferTest") {
@Override
public void run() {
try {
// should be much larger than the lock acquisition timeout
Thread.sleep(1000);
manager(c2).stop();
// stLock.unblockNewTransactions(1000);
} catch (InterruptedException e) {
log.errorf(e, "Error stopping cache");
}
}
};
worker.start();
try {
// finally commit or rollback the transaction
if (commit) {
txCoordinator.commit(localTx, false);
} else {
txCoordinator.rollback(localTx);
}
// make the transaction manager forget about our tx so that we don't get rollback exceptions in the log
tm(c1).suspend();
} finally {