SGF-598 - Add support for suspend and resume in GemfireTransactionManager.
(cherry picked from commit 7c594e7cbad6854ffaf3b162647ea9eced0b3d92) Signed-off-by: John Blum <jblum@pivotal.io>
This commit is contained in:
John Blum
@@ -17,9 +17,15 @@
|
||||
|
||||
package org.springframework.data.gemfire.transaction;
|
||||
|
||||
import org.apache.geode.cache.Cache;
|
||||
import static org.springframework.data.gemfire.transaction.GemfireTransactionManager.CacheHolder.newCacheHolder;
|
||||
import static org.springframework.data.gemfire.transaction.GemfireTransactionManager.CacheTransactionObject.newCacheTransactionObject;
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.apache.geode.cache.CacheTransactionManager;
|
||||
import org.apache.geode.cache.GemFireCache;
|
||||
import org.apache.geode.cache.Region;
|
||||
import org.apache.geode.cache.TransactionId;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.transaction.CannotCreateTransactionException;
|
||||
import org.springframework.transaction.NoTransactionException;
|
||||
@@ -33,252 +39,380 @@ import org.springframework.transaction.support.TransactionSynchronizationManager
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Local transaction manager for GemFire Enterprise Fabric (GEF). Provides a
|
||||
* {@link PlatformTransactionManager} implementation for a single GemFire
|
||||
* {@link CacheTransactionManager}.
|
||||
* Local Transaction Management for Pivotal GemFire. Provides a Spring {@link PlatformTransactionManager} implementation
|
||||
* for the Pivotal GemFire {@link CacheTransactionManager}.
|
||||
*
|
||||
* Binds one or multiple GemFire regions for the specified {@link Cache} to the
|
||||
* thread, potentially allowing for one region per cache model.
|
||||
* Binds one or multiple GemFire {@link Region Regions} for the specified {@link GemFireCache} to the thread,
|
||||
* potentially allowing for one {@link Region} per {@link GemFireCache} model.
|
||||
*
|
||||
* <p>
|
||||
* This local strategy is an alternative to executing cache operations within
|
||||
* JTA transactions. Its advantage is that is able to work in any environment,
|
||||
* for example a stand-alone application or a test suite. It is <i>not</i> able
|
||||
* to provide XA transactions, for example to share transactions with data
|
||||
* access.
|
||||
* This local strategy is an alternative to executing cache operations within JTA transactions.
|
||||
* Its advantage is that is able to work in any environment, for example a stand-alone application
|
||||
* or a test suite. It is <i>not</i> able to provide XA transactions, for example to share transactions
|
||||
* with data access.
|
||||
*
|
||||
* <p>
|
||||
* To prevent dirty reads, by default, the cache is configured to return copies
|
||||
* rather then direct references for <code>get</code> operations. As a
|
||||
* workaround, one could use explicitly deep copy objects before making changes
|
||||
* to them to avoid unnecessary copying on every fetch.
|
||||
* By default, to prevent dirty reads, the {@link GemFireCache} is configured to return copies rather then direct references
|
||||
* for <code>get</code> data access operations. As a workaround, one could use explicitly deep copy objects before
|
||||
* making changes to them to avoid unnecessary copying on every fetch.
|
||||
*
|
||||
* @author Costin Leau
|
||||
* @author John Blum
|
||||
* @see org.apache.geode.cache.Cache#setCopyOnRead(boolean)
|
||||
* @see org.apache.geode.cache.GemFireCache#setCopyOnRead(boolean)
|
||||
* @see org.apache.geode.cache.CacheTransactionManager
|
||||
* @see org.apache.geode.cache.Region#get(Object)
|
||||
* @see org.apache.geode.CopyHelper#copy(Object)
|
||||
* @see org.springframework.transaction.support.AbstractPlatformTransactionManager
|
||||
* @see #setCopyOnRead(boolean)
|
||||
*/
|
||||
// TODO add lenient behavior if a transaction is already started on the current
|
||||
// thread (what should happen then)
|
||||
@SuppressWarnings("serial")
|
||||
@SuppressWarnings("unused")
|
||||
public class GemfireTransactionManager extends AbstractPlatformTransactionManager
|
||||
implements InitializingBean, ResourceTransactionManager {
|
||||
|
||||
private Cache cache;
|
||||
protected static final TimeUnit DEFAULT_RESUME_WAIT_TIME_UNIT = TimeUnit.SECONDS;
|
||||
|
||||
private GemFireCache cache;
|
||||
|
||||
private boolean copyOnRead = true;
|
||||
|
||||
private Long resumeWaitTime;
|
||||
|
||||
private TimeUnit resumeWaitTimeUnit = DEFAULT_RESUME_WAIT_TIME_UNIT;
|
||||
|
||||
/**
|
||||
* Creates a new GemfireTransactionManager instance.
|
||||
* Constructs an instance of the {@link GemfireTransactionManager}.
|
||||
*/
|
||||
public GemfireTransactionManager() {
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new GemfireTransactionManager instance.
|
||||
* Constructs an instance of the {@link GemfireTransactionManager} initialized with
|
||||
* the given {@link GemFireCache} reference.
|
||||
*
|
||||
* @param cache a reference to the GemFire Cache associated with Cache transactions.
|
||||
* @param cache reference to the {@link GemFireCache} associated with cache transactions.
|
||||
* @see org.apache.geode.cache.GemFireCache
|
||||
* @see #afterPropertiesSet()
|
||||
*/
|
||||
public GemfireTransactionManager(Cache cache) {
|
||||
public GemfireTransactionManager(GemFireCache cache) {
|
||||
this.cache = cache;
|
||||
afterPropertiesSet();
|
||||
}
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
public void afterPropertiesSet() {
|
||||
Assert.notNull(cache, "Cache property is required");
|
||||
cache.setCopyOnRead(copyOnRead);
|
||||
Assert.notNull(this.cache, "Cache is required");
|
||||
this.cache.setCopyOnRead(isCopyOnRead());
|
||||
}
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
protected Object doGetTransaction() throws TransactionException {
|
||||
CacheTransactionObject txObject = new CacheTransactionObject();
|
||||
CacheHolder cacheHolder = (CacheHolder) TransactionSynchronizationManager.getResource(getCache());
|
||||
txObject.setHolder(cacheHolder);
|
||||
return txObject;
|
||||
return newCacheTransactionObject((CacheHolder) TransactionSynchronizationManager.getResource(getCache()));
|
||||
}
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
protected boolean isExistingTransaction(Object transaction) throws TransactionException {
|
||||
CacheTransactionObject txObject = (CacheTransactionObject) transaction;
|
||||
// Consider a pre-bound cache as transaction.
|
||||
return (txObject.getHolder() != null);
|
||||
// consider a pre-bound cache as a transaction
|
||||
return ((CacheTransactionObject) transaction).isHolding();
|
||||
}
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionException {
|
||||
CacheTransactionObject txObject = (CacheTransactionObject) transaction;
|
||||
|
||||
Cache cache;
|
||||
|
||||
try {
|
||||
cache = getCache();
|
||||
CacheTransactionObject cacheTransaction = (CacheTransactionObject) transaction;
|
||||
GemFireCache cache = getCache();
|
||||
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Acquired Cache [" + cache + "] for local Cache transaction");
|
||||
logger.debug(String.format("Acquired GemFire Cache [%s] for local cache transaction", cache));
|
||||
}
|
||||
|
||||
txObject.setHolder(new CacheHolder());
|
||||
cache.getCacheTransactionManager().begin();
|
||||
TransactionSynchronizationManager.bindResource(cache, txObject.getHolder());
|
||||
}
|
||||
CacheTransactionManager cacheTransactionManager = getCacheTransactionManager();
|
||||
|
||||
catch (IllegalStateException ex) {
|
||||
throw new CannotCreateTransactionException(
|
||||
"An ongoing transaction already is already associated with the current thread; are there multiple transaction managers ?",
|
||||
ex);
|
||||
// begin GemFire local cache transaction
|
||||
cacheTransactionManager.begin();
|
||||
|
||||
TransactionId transactionId = cacheTransactionManager.getTransactionId();
|
||||
|
||||
if (transactionId != null) {
|
||||
TransactionSynchronizationManager.bindResource(cache,
|
||||
cacheTransaction.setAndGetHolder(newCacheHolder(transactionId)));
|
||||
}
|
||||
}
|
||||
catch (IllegalStateException e) {
|
||||
throw new CannotCreateTransactionException(String.format("%1$s; %2$s",
|
||||
"An existing, ongoing transaction is already associated with the current thread",
|
||||
"are multiple transaction managers present?"), e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
|
||||
// CacheTransactionObject txObject = (CacheTransactionObject)
|
||||
// status.getTransaction();
|
||||
if (status.isDebug()) {
|
||||
logger.debug("Committing Gemfire local transaction on Cache [" + cache + "]");
|
||||
}
|
||||
try {
|
||||
cache.getCacheTransactionManager().commit();
|
||||
}
|
||||
catch (IllegalStateException ex) {
|
||||
throw new NoTransactionException(
|
||||
"No transaction associated with the current thread; are there multiple transaction managers ?", ex);
|
||||
}
|
||||
if (status.isDebug()) {
|
||||
logger.debug("Committing local cache transaction");
|
||||
}
|
||||
|
||||
catch (org.apache.geode.cache.TransactionException ex) {
|
||||
throw new GemfireTransactionCommitException("Unexpected failure on commit of Cache local transaction", ex);
|
||||
getCacheTransactionManager().commit();
|
||||
}
|
||||
catch (IllegalStateException e) {
|
||||
throw new NoTransactionException(
|
||||
"No transaction is associated with the current thread; are multiple transaction managers present?", e);
|
||||
}
|
||||
catch (org.apache.geode.cache.TransactionException e) {
|
||||
throw new GemfireTransactionCommitException(
|
||||
"Unexpected failure occurred on commit of local cache transaction", e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
protected Object doSuspend(Object transaction) throws TransactionException {
|
||||
if (getCacheTransactionManager().suspend() != null) {
|
||||
TransactionSynchronizationManager.unbindResource(getCache());
|
||||
return ((CacheTransactionObject) transaction).setAndReturnExistingHolder(null);
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
protected void doResume(Object transaction, Object suspendedResources) throws TransactionException {
|
||||
if (suspendedResources instanceof CacheHolder) {
|
||||
CacheHolder holder = (CacheHolder) suspendedResources;
|
||||
|
||||
boolean resumeSuccessful = (isResumeWaitTimeSet()
|
||||
? getCacheTransactionManager().tryResume(holder.getTransactionId(),
|
||||
getResumeWaitTime(), getResumeWaitTimeUnit())
|
||||
: getCacheTransactionManager().tryResume(holder.getTransactionId()));
|
||||
|
||||
if (resumeSuccessful) {
|
||||
TransactionSynchronizationManager.bindResource(getCache(),
|
||||
((CacheTransactionObject) transaction).setAndGetHolder(holder));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
protected void doRollback(DefaultTransactionStatus status) throws TransactionException {
|
||||
if (status.isDebug()) {
|
||||
logger.debug("Rolling back Cache local transaction for [" + cache + "]");
|
||||
}
|
||||
try {
|
||||
cache.getCacheTransactionManager().rollback();
|
||||
if (status.isDebug()) {
|
||||
logger.debug("Rolling back local cache transaction");
|
||||
}
|
||||
|
||||
getCacheTransactionManager().rollback();
|
||||
}
|
||||
catch (IllegalStateException ex) {
|
||||
catch (IllegalStateException e) {
|
||||
throw new NoTransactionException(
|
||||
"No transaction associated with the current thread; are there multiple transaction managers ?", ex);
|
||||
"No transaction is associated with the current thread; are multiple transaction managers present?",
|
||||
e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doSetRollbackOnly(DefaultTransactionStatus status) {
|
||||
CacheTransactionObject txObject = (CacheTransactionObject) status.getTransaction();
|
||||
if (status.isDebug()) {
|
||||
logger.debug("Setting Gemfire local transaction [" + txObject.getHolder() + "] rollback-only");
|
||||
}
|
||||
txObject.getHolder().setRollbackOnly();
|
||||
}
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
protected void doCleanupAfterCompletion(Object transaction) {
|
||||
// Remove the cache holder from the thread.
|
||||
TransactionSynchronizationManager.unbindResource(cache);
|
||||
TransactionSynchronizationManager.unbindResource(getCache());
|
||||
}
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
protected void doSetRollbackOnly(DefaultTransactionStatus status) {
|
||||
((CacheTransactionObject) status.getTransaction()).getHolder().setRollbackOnly();
|
||||
}
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
protected final boolean useSavepointForNestedTransaction() {
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the Cache that this instance manages local transactions for.
|
||||
*
|
||||
* @return Gemfire cache
|
||||
* Sets a reference to the {@link GemFireCache} for which this transaction manager
|
||||
* will manage local cache transactions.
|
||||
* @param cache reference to the {@link GemFireCache}.
|
||||
* @see org.apache.geode.cache.GemFireCache
|
||||
*/
|
||||
public Cache getCache() {
|
||||
return cache;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the Cache that this instance manages local transactions for.
|
||||
*
|
||||
* @param cache Gemfire cache
|
||||
*/
|
||||
public void setCache(Cache cache) {
|
||||
public void setCache(GemFireCache cache) {
|
||||
this.cache = cache;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object getResourceFactory() {
|
||||
return getCache();
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the GemFire Cache {@link Region} (as an alternative in setting in the Cache directly).
|
||||
* Returns a reference to the {@link GemFireCache} for which this transaction manager
|
||||
* will manage local cache transactions.
|
||||
*
|
||||
* @param <K> the Region key class type.
|
||||
* @param <V> the Region value class type.
|
||||
* @param region the Gemfire Cache Region directly involved in the Cache transaction.
|
||||
* @return a reference to the {@link GemFireCache}.
|
||||
* @see org.apache.geode.cache.GemFireCache
|
||||
*/
|
||||
public <K, V> void setRegion(Region<K, V> region) {
|
||||
Assert.notNull(region, "non-null arguments are required");
|
||||
this.cache = (Cache)region.getRegionService();
|
||||
public GemFireCache getCache() {
|
||||
return this.cache;
|
||||
}
|
||||
|
||||
protected CacheTransactionManager getCacheTransactionManager() {
|
||||
return getCache().getCacheTransactionManager();
|
||||
}
|
||||
|
||||
/**
|
||||
* Indicates whether the cache returns direct references or copies of the
|
||||
* objects (default) it manages. While copies imply additional work for
|
||||
* every fetch operation, direct references can cause dirty reads across
|
||||
* concurrent threads in the same VM, whether or not transactions are used.
|
||||
* Set whether the cache returns direct object references or copies of the objects it manages.
|
||||
* While copies imply additional work for every fetch operation, direct object references can
|
||||
* cause dirty reads across concurrent threads in the same VM, whether or not transactions are used.
|
||||
*
|
||||
* One could explicitly deep copy objects before making changes (for example
|
||||
* by using {@link org.apache.geode.CopyHelper#copy(Object)} in which
|
||||
* case this setting can be set to <code>false</code>. However, unless there
|
||||
* is a measurable performance penalty, the recommendation is to keep this
|
||||
* setting to <code>true</code>
|
||||
* One could explicitly deep copy objects before making changes (for example by using
|
||||
* {@link org.apache.geode.CopyHelper#copy(Object)} in which case this setting
|
||||
* can be set to <code>false</code>
|
||||
*
|
||||
* @param copyOnRead whether copies (default) rather then direct references
|
||||
* will be returned on fetch operations
|
||||
* However, unless there is a measurable performance penalty, the recommendation is
|
||||
* to keep this setting to <code>true</code>.
|
||||
*
|
||||
* @param copyOnRead boolean value indicating whether copies (default) rather then direct object references
|
||||
* will be returned on fetch operations.
|
||||
*/
|
||||
public void setCopyOnRead(boolean copyOnRead) {
|
||||
this.copyOnRead = copyOnRead;
|
||||
}
|
||||
|
||||
/**
|
||||
* Indicates whether copy on read is set or not on the transaction manager.
|
||||
* Indicates whether copy on read is set and used for fetch data access operations.
|
||||
*
|
||||
* @return the setting for copy-on-read.
|
||||
* @see #setCopyOnRead(boolean)
|
||||
* @return the copyOnRead
|
||||
*/
|
||||
public boolean isCopyOnRead() {
|
||||
return copyOnRead;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the GemFire cache {@link Region} as an alternative in setting in the {@link GemFireCache} directly.
|
||||
*
|
||||
* @param <K> {@link Class} type of the {@link Region} key.
|
||||
* @param <V> {@link Class} type of the {@link Region} value.
|
||||
* @param region GemFire cache {@link Region} directly involved in the local cache transaction.
|
||||
* @throws IllegalArgumentException if {@link Region} is {@literal null}.
|
||||
* @see org.apache.geode.cache.Region
|
||||
*/
|
||||
public <K, V> void setRegion(Region<K, V> region) {
|
||||
Assert.notNull(region, "Region must not be null");
|
||||
this.cache = (GemFireCache) region.getRegionService();
|
||||
}
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
public Object getResourceFactory() {
|
||||
return getCache();
|
||||
}
|
||||
|
||||
public void setResumeWaitTime(Long resumeWaitTime) {
|
||||
this.resumeWaitTime = resumeWaitTime;
|
||||
}
|
||||
|
||||
protected Long getResumeWaitTime() {
|
||||
return this.resumeWaitTime;
|
||||
}
|
||||
|
||||
protected boolean isResumeWaitTimeSet() {
|
||||
Long resumeWaitTime = getResumeWaitTime();
|
||||
return (resumeWaitTime != null && resumeWaitTime > 0);
|
||||
}
|
||||
|
||||
public void setResumeWaitTimeUnit(TimeUnit resumeWaitTimeUnit) {
|
||||
this.resumeWaitTimeUnit = resumeWaitTimeUnit;
|
||||
}
|
||||
|
||||
protected TimeUnit getResumeWaitTimeUnit() {
|
||||
TimeUnit localResumeWaitTimeUnit = this.resumeWaitTimeUnit;
|
||||
return (localResumeWaitTimeUnit != null ? localResumeWaitTimeUnit : DEFAULT_RESUME_WAIT_TIME_UNIT);
|
||||
}
|
||||
|
||||
/**
|
||||
* GemfireTM local transaction object.
|
||||
*
|
||||
* @author Costin Leau
|
||||
*/
|
||||
private static class CacheTransactionObject {
|
||||
protected static class CacheTransactionObject {
|
||||
|
||||
private CacheHolder cacheHolder;
|
||||
|
||||
public CacheHolder getHolder() {
|
||||
return cacheHolder;
|
||||
static CacheTransactionObject newCacheTransactionObject(CacheHolder cacheHolder) {
|
||||
CacheTransactionObject transactionObject = new CacheTransactionObject();
|
||||
transactionObject.setHolder(cacheHolder);
|
||||
return transactionObject;
|
||||
}
|
||||
|
||||
public void setHolder(CacheHolder cacheHolder) {
|
||||
this.cacheHolder = cacheHolder;
|
||||
boolean isHolding() {
|
||||
return (getHolder() != null);
|
||||
}
|
||||
|
||||
CacheHolder getHolder() {
|
||||
return this.cacheHolder;
|
||||
}
|
||||
|
||||
void setHolder(CacheHolder holder) {
|
||||
this.cacheHolder = holder;
|
||||
}
|
||||
|
||||
CacheHolder setAndGetHolder(CacheHolder holder) {
|
||||
setHolder(holder);
|
||||
return getHolder();
|
||||
}
|
||||
|
||||
CacheHolder setAndReturnExistingHolder(CacheHolder cacheHolder) {
|
||||
CacheHolder existingHolder = getHolder();
|
||||
setHolder(cacheHolder);
|
||||
return existingHolder;
|
||||
}
|
||||
}
|
||||
|
||||
private static class CacheHolder {
|
||||
protected static class CacheHolder {
|
||||
|
||||
private boolean rollbackOnly = false;
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public boolean isRollbackOnly() {
|
||||
return rollbackOnly;
|
||||
private TransactionId transactionId;
|
||||
|
||||
static CacheHolder newCacheHolder(TransactionId transactionId) {
|
||||
CacheHolder cacheHolder = new CacheHolder();
|
||||
cacheHolder.transactionId = transactionId;
|
||||
return cacheHolder;
|
||||
}
|
||||
|
||||
public void setRollbackOnly() {
|
||||
rollbackOnly = true;
|
||||
public boolean isRollbackOnly() {
|
||||
return this.rollbackOnly;
|
||||
}
|
||||
|
||||
void setRollbackOnly() {
|
||||
this.rollbackOnly = true;
|
||||
}
|
||||
|
||||
TransactionId getTransactionId() {
|
||||
return this.transactionId;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user