Hibernate
/
hibernate-orm
mirror of https://github.com/hibernate/hibernate-orm
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

HHH-7910 Transaction timeout can cause non-threadsafe session access by reaper thread

This commit is contained in:
Brett Meyer 2013-02-19 11:31:30 -05:00
parent 42f3422720
commit b6d740d6d2
9 changed files with 205 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
hibernate-core/src/main/java/org/hibernate/cfg/AvailableSettings.java
View File

@ -623,4 +623,15 @@ public interface AvailableSettings {
public static final String USE_DIRECT_REFERENCE_CACHE_ENTRIES = "hibernate.cache.use_reference_entries";
public static final String USE_NATIONALIZED_CHARACTER_DATA = "hibernate.use_nationalized_character_data";
/**
* A transaction can be rolled back by another thread ("tracking by thread")
* -- not the original application. Examples of this include a JTA
* transaction timeout handled by a background reaper thread. The ability
* to handle this situation requires checking the Thread ID every time
* Session is called. This can certainly have performance considerations.
*
* Default is true (enabled).
*/
public static final String JTA_TRACK_BY_THREAD = "hibernate.jta.track_by_thread";
}

10
hibernate-core/src/main/java/org/hibernate/cfg/Settings.java
View File

@ -96,6 +96,8 @@ public final class Settings {
private MultiTableBulkIdStrategy multiTableBulkIdStrategy;
private BatchFetchStyle batchFetchStyle;
private boolean directReferenceCacheEntriesEnabled;
private boolean jtaTrackByThread;
/**
@ -508,4 +510,12 @@ public final class Settings {
void setDefaultNullPrecedence(NullPrecedence defaultNullPrecedence) {
this.defaultNullPrecedence = defaultNullPrecedence;
}
public boolean isJtaTrackByThread() {
return jtaTrackByThread;
}
public void setJtaTrackByThread(boolean jtaTrackByThread) {
this.jtaTrackByThread = jtaTrackByThread;
}
}

10
hibernate-core/src/main/java/org/hibernate/cfg/SettingsFactory.java
View File

@ -395,6 +395,16 @@ public class SettingsFactory implements Serializable {
}
settings.setInitializeLazyStateOutsideTransactions( initializeLazyStateOutsideTransactionsEnabled );
boolean jtaTrackByThread = ConfigurationHelper.getBoolean(
AvailableSettings.JTA_TRACK_BY_THREAD,
properties,
true
);
if ( debugEnabled ) {
LOG.debugf( "JTA Track by Thread: %s", enabledDisabled(jtaTrackByThread) );
}
settings.setJtaTrackByThread( jtaTrackByThread );
return settings;
}

1
hibernate-core/src/main/java/org/hibernate/engine/transaction/internal/TransactionCoordinatorImpl.java
View File

@ -267,6 +267,7 @@ public class TransactionCoordinatorImpl implements TransactionCoordinator {
}
public void pulse() {
getSynchronizationCallbackCoordinator().pulse();
if ( transactionFactory().compatibleWithJtaSynchronization() ) {
// the configured transaction strategy says it supports callbacks via JTA synchronization, so attempt to
// register JTA synchronization if possible

74
hibernate-core/src/main/java/org/hibernate/engine/transaction/synchronization/internal/SynchronizationCallbackCoordinatorImpl.java
View File

@ -25,9 +25,8 @@ package org.hibernate.engine.transaction.synchronization.internal;
import javax.transaction.SystemException;
import org.jboss.logging.Logger;
import org.hibernate.TransactionException;
import org.hibernate.cfg.Settings;
import org.hibernate.engine.transaction.internal.jta.JtaStatusHelper;
import org.hibernate.engine.transaction.spi.TransactionContext;
import org.hibernate.engine.transaction.spi.TransactionCoordinator;
@ -36,31 +35,43 @@ import org.hibernate.engine.transaction.synchronization.spi.ExceptionMapper;
import org.hibernate.engine.transaction.synchronization.spi.ManagedFlushChecker;
import org.hibernate.engine.transaction.synchronization.spi.SynchronizationCallbackCoordinator;
import org.hibernate.internal.CoreMessageLogger;
import org.jboss.logging.Logger;
/**
* Manages callbacks from the {@link javax.transaction.Synchronization} registered by Hibernate.
*
*
* @author Steve Ebersole
* @author Brett Meyer
*/
public class SynchronizationCallbackCoordinatorImpl implements SynchronizationCallbackCoordinator {
private static final CoreMessageLogger LOG = Logger.getMessageLogger( CoreMessageLogger.class, SynchronizationCallbackCoordinatorImpl.class.getName() );
private static final CoreMessageLogger LOG = Logger.getMessageLogger( CoreMessageLogger.class,
SynchronizationCallbackCoordinatorImpl.class.getName() );
private final TransactionCoordinator transactionCoordinator;
private final Settings settings;
private ManagedFlushChecker managedFlushChecker;
private AfterCompletionAction afterCompletionAction;
private ExceptionMapper exceptionMapper;
private long registrationThreadId;
private final int NO_STATUS = -1;
private int delayedCompletionHandlingStatus;
public SynchronizationCallbackCoordinatorImpl(TransactionCoordinator transactionCoordinator) {
this.transactionCoordinator = transactionCoordinator;
this.settings = transactionCoordinator.getTransactionContext()
.getTransactionEnvironment().getSessionFactory().getSettings();
reset();
pulse();
}
public void reset() {
managedFlushChecker = STANDARD_MANAGED_FLUSH_CHECKER;
exceptionMapper = STANDARD_EXCEPTION_MAPPER;
afterCompletionAction = STANDARD_AFTER_COMPLETION_ACTION;
delayedCompletionHandlingStatus = NO_STATUS;
}
@Override
@ -78,7 +89,6 @@ public class SynchronizationCallbackCoordinatorImpl implements SynchronizationCa
this.afterCompletionAction = afterCompletionAction;
}
// sync callbacks ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
public void beforeCompletion() {
@ -86,16 +96,14 @@ public class SynchronizationCallbackCoordinatorImpl implements SynchronizationCa
boolean flush;
try {
final int status = transactionCoordinator
.getTransactionContext()
.getTransactionEnvironment()
.getJtaPlatform()
.getCurrentStatus();
final int status = transactionCoordinator.getTransactionContext().getTransactionEnvironment()
.getJtaPlatform().getCurrentStatus();
flush = managedFlushChecker.shouldDoManagedFlush( transactionCoordinator, status );
}
catch ( SystemException se ) {
setRollbackOnly();
throw exceptionMapper.mapStatusCheckFailure( "could not determine transaction status in beforeCompletion()", se );
throw exceptionMapper.mapStatusCheckFailure(
"could not determine transaction status in beforeCompletion()", se );
}
try {
@ -119,6 +127,35 @@ public class SynchronizationCallbackCoordinatorImpl implements SynchronizationCa
}
public void afterCompletion(int status) {
if ( !settings.isJtaTrackByThread() || isRegistrationThread() ) {
doAfterCompletion( status );
}
else if ( JtaStatusHelper.isRollback( status ) ) {
// The transaction was rolled back by another thread -- not the
// original application. Examples of this include a JTA transaction
// timeout getting cleaned up by a reaper thread. If this happens,
// afterCompletion must be handled by the original thread in order
// to prevent non-threadsafe session use. Set the flag here and
// check for it in SessionImpl. See HHH-7910.
LOG.warnv( "Transaction afterCompletion called by a background thread! Delaying action until the original thread can handle it. [status={0}]", status );
delayedCompletionHandlingStatus = status;
}
}
public void pulse() {
if ( settings.isJtaTrackByThread() ) {
registrationThreadId = Thread.currentThread().getId();
}
}
public void delayedAfterCompletion() {
if ( delayedCompletionHandlingStatus != NO_STATUS ) {
doAfterCompletion( delayedCompletionHandlingStatus );
delayedCompletionHandlingStatus = NO_STATUS;
}
}
private void doAfterCompletion(int status) {
LOG.tracev( "Transaction after completion callback [status={0}]", status );
try {
@ -134,6 +171,10 @@ public class SynchronizationCallbackCoordinatorImpl implements SynchronizationCa
}
}
private boolean isRegistrationThread() {
return Thread.currentThread().getId() == registrationThreadId;
}
private TransactionContext transactionContext() {
return transactionCoordinator.getTransactionContext();
}
@ -141,17 +182,18 @@ public class SynchronizationCallbackCoordinatorImpl implements SynchronizationCa
private static final ManagedFlushChecker STANDARD_MANAGED_FLUSH_CHECKER = new ManagedFlushChecker() {
@Override
public boolean shouldDoManagedFlush(TransactionCoordinator coordinator, int jtaStatus) {
return ! coordinator.getTransactionContext().isClosed() &&
! coordinator.getTransactionContext().isFlushModeNever() &&
coordinator.getTransactionContext().isFlushBeforeCompletionEnabled() &&
! JtaStatusHelper.isRollback( jtaStatus );
return !coordinator.getTransactionContext().isClosed()
&& !coordinator.getTransactionContext().isFlushModeNever()
&& coordinator.getTransactionContext().isFlushBeforeCompletionEnabled()
&& !JtaStatusHelper.isRollback( jtaStatus );
}
};
private static final ExceptionMapper STANDARD_EXCEPTION_MAPPER = new ExceptionMapper() {
public RuntimeException mapStatusCheckFailure(String message, SystemException systemException) {
LOG.error( LOG.unableToDetermineTransactionStatus(), systemException );
return new TransactionException( "could not determine transaction status in beforeCompletion()", systemException );
return new TransactionException( "could not determine transaction status in beforeCompletion()",
systemException );
}
public RuntimeException mapManagedFlushFailure(String message, RuntimeException failure) {

2
hibernate-core/src/main/java/org/hibernate/engine/transaction/synchronization/spi/SynchronizationCallbackCoordinator.java
View File

@ -31,5 +31,7 @@ import javax.transaction.Synchronization;
public interface SynchronizationCallbackCoordinator extends Synchronization{
public void setManagedFlushChecker(ManagedFlushChecker managedFlushChecker);
public void setAfterCompletionAction(AfterCompletionAction afterCompletionAction);
public void pulse();
public void delayedAfterCompletion();
public void setExceptionMapper(ExceptionMapper exceptionMapper);
}

52
hibernate-core/src/main/java/org/hibernate/internal/SessionImpl.java
View File

@ -170,6 +170,7 @@ import org.jboss.logging.Logger;
*
* @author Gavin King
* @author Steve Ebersole
* @author Brett Meyer
*/
public final class SessionImpl extends AbstractSessionImpl implements EventSource {
@ -321,7 +322,9 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
public void clear() {
errorIfClosed();
checkTransactionSynchStatus();
// Do not call checkTransactionSynchStatus() here -- if a delayed
// afterCompletion exists, it can cause an infinite loop.
pulseTransactionCoordinator();
internalClear();
}
@ -707,6 +710,11 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
if ( persistenceContext.getCascadeLevel() == 0 ) {
actionQueue.checkNoUnresolvedActionsAfterOperation();
}
delayedAfterCompletion();
}
private void delayedAfterCompletion() {
transactionCoordinator.getSynchronizationCallbackCoordinator().delayedAfterCompletion();
}
// saveOrUpdate() operations ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@ -809,6 +817,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
for ( LockEventListener listener : listeners( EventType.LOCK ) ) {
listener.onLock( event );
}
delayedAfterCompletion();
}
@ -833,6 +842,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
for ( PersistEventListener listener : listeners( EventType.PERSIST ) ) {
listener.onPersist( event, copiedAlready );
}
delayedAfterCompletion();
}
private void firePersist(PersistEvent event) {
@ -868,6 +878,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
for ( PersistEventListener listener : listeners( EventType.PERSIST_ONFLUSH ) ) {
listener.onPersist( event, copiedAlready );
}
delayedAfterCompletion();
}
private void firePersistOnFlush(PersistEvent event) {
@ -912,6 +923,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
for ( MergeEventListener listener : listeners( EventType.MERGE ) ) {
listener.onMerge( event, copiedAlready );
}
delayedAfterCompletion();
}
@ -944,6 +956,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
for ( DeleteEventListener listener : listeners( EventType.DELETE ) ) {
listener.onDelete( event );
}
delayedAfterCompletion();
}
private void fireDelete(DeleteEvent event, Set transientEntities) {
@ -952,6 +965,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
for ( DeleteEventListener listener : listeners( EventType.DELETE ) ) {
listener.onDelete( event, transientEntities );
}
delayedAfterCompletion();
}
@ -1077,6 +1091,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
for ( LoadEventListener listener : listeners( EventType.LOAD ) ) {
listener.onLoad( event, loadType );
}
delayedAfterCompletion();
}
private void fireResolveNaturalId(ResolveNaturalIdEvent event) {
@ -1085,6 +1100,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
for ( ResolveNaturalIdEventListener listener : listeners( EventType.RESOLVE_NATURAL_ID ) ) {
listener.onResolveNaturalId( event );
}
delayedAfterCompletion();
}
@ -1121,6 +1137,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
for ( RefreshEventListener listener : listeners( EventType.REFRESH ) ) {
listener.onRefresh( event );
}
delayedAfterCompletion();
}
private void fireRefresh(Map refreshedAlready, RefreshEvent event) {
@ -1129,6 +1146,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
for ( RefreshEventListener listener : listeners( EventType.REFRESH ) ) {
listener.onRefresh( event, refreshedAlready );
}
delayedAfterCompletion();
}
@ -1149,6 +1167,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
for ( ReplicateEventListener listener : listeners( EventType.REPLICATE ) ) {
listener.onReplicate( event );
}
delayedAfterCompletion();
}
@ -1168,6 +1187,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
for ( EvictEventListener listener : listeners( EventType.EVICT ) ) {
listener.onEvict( event );
}
delayedAfterCompletion();
}
/**
@ -1199,6 +1219,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
for ( DirtyCheckEventListener listener : listeners( EventType.DIRTY_CHECK ) ) {
listener.onDirtyCheck( event );
}
delayedAfterCompletion();
return event.isDirty();
}
@ -1212,6 +1233,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
for ( FlushEventListener listener : listeners( EventType.FLUSH ) ) {
listener.onFlush( flushEvent );
}
delayedAfterCompletion();
}
public void forceFlush(EntityEntry entityEntry) throws HibernateException {
@ -1250,6 +1272,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
finally {
dontFlushFromFind--;
afterOperation(success);
delayedAfterCompletion();
}
return results;
}
@ -1269,6 +1292,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
}
finally {
afterOperation(success);
delayedAfterCompletion();
}
return result;
}
@ -1290,6 +1314,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
success = true;
} finally {
afterOperation(success);
delayedAfterCompletion();
}
return result;
}
@ -1306,6 +1331,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
return plan.performIterate( queryParameters, this );
}
finally {
delayedAfterCompletion();
dontFlushFromFind--;
}
}
@ -1320,6 +1346,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
return plan.performScroll( queryParameters, this );
}
finally {
delayedAfterCompletion();
dontFlushFromFind--;
}
}
@ -1334,13 +1361,16 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
getFilterQueryPlan( collection, queryString, null, false ).getParameterMetadata()
);
filter.setComment( queryString );
delayedAfterCompletion();
return filter;
}
public Query getNamedQuery(String queryName) throws MappingException {
errorIfClosed();
checkTransactionSynchStatus();
return super.getNamedQuery( queryName );
Query query = super.getNamedQuery( queryName );
delayedAfterCompletion();
return query;
}
public Object instantiate(String entityName, Serializable id) throws HibernateException {
@ -1357,6 +1387,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
if ( result == null ) {
result = persister.instantiate( id, this );
}
delayedAfterCompletion();
return result;
}
@ -1526,6 +1557,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
finally {
dontFlushFromFind--;
afterOperation(success);
delayedAfterCompletion();
}
return results;
}
@ -1535,7 +1567,9 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
errorIfClosed();
checkTransactionSynchStatus();
FilterQueryPlan plan = getFilterQueryPlan( collection, filter, queryParameters, true );
return plan.performIterate( queryParameters, this );
Iterator itr = plan.performIterate( queryParameters, this );
delayedAfterCompletion();
return itr;
}
public Criteria createCriteria(Class persistentClass, String alias) {
@ -1582,6 +1616,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
return loader.scroll(this, scrollMode);
}
finally {
delayedAfterCompletion();
dontFlushFromFind--;
}
}
@ -1633,6 +1668,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
finally {
dontFlushFromFind--;
afterOperation(success);
delayedAfterCompletion();
}
return results;
@ -1732,6 +1768,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
// an entry in the session's persistence context and the entry reports
// that the entity has not been removed
EntityEntry entry = persistenceContext.getEntry( object );
delayedAfterCompletion();
return entry != null && entry.getStatus() != Status.DELETED && entry.getStatus() != Status.GONE;
}
@ -1786,6 +1823,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
return loader.scroll(queryParameters, this);
}
finally {
delayedAfterCompletion();
dontFlushFromFind--;
}
}
@ -1813,6 +1851,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
}
finally {
dontFlushFromFind--;
delayedAfterCompletion();
afterOperation(success);
}
}
@ -1830,6 +1869,7 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
for ( InitializeCollectionEventListener listener : listeners( EventType.INIT_COLLECTION ) ) {
listener.onInitializeCollection( event );
}
delayedAfterCompletion();
}
public String bestGuessEntityName(Object object) {
@ -2085,8 +2125,12 @@ public final class SessionImpl extends AbstractSessionImpl implements EventSourc
loadQueryInfluencers.disableFetchProfile( name );
}
private void checkTransactionSynchStatus() {
pulseTransactionCoordinator();
delayedAfterCompletion();
}
private void pulseTransactionCoordinator() {
if ( !isClosed() ) {
transactionCoordinator.pulse();
}

7
hibernate-entitymanager/src/test/java/org/hibernate/jpa/test/transaction/Book.java
View File

@ -18,6 +18,13 @@ public class Book {
@Version
public Integer version;
public Book() {}
public Book(String name, Integer version) {
this.name = name;
this.version = version;
}
public Integer getId() {
return id;

67
hibernate-entitymanager/src/test/java/org/hibernate/jpa/test/transaction/TransactionJoiningTest.java
View File

@ -23,25 +23,29 @@
*/
package org.hibernate.jpa.test.transaction;
import javax.persistence.EntityManager;
import javax.transaction.Synchronization;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import javax.persistence.EntityManager;
import javax.transaction.Status;
import javax.transaction.Synchronization;
import org.hibernate.Session;
import org.hibernate.Transaction;
import org.hibernate.jpa.test.BaseEntityManagerFunctionalTestCase;
import org.hibernate.engine.spi.SessionImplementor;
import org.hibernate.engine.transaction.internal.jta.CMTTransaction;
import org.hibernate.engine.transaction.internal.jta.JtaStatusHelper;
import org.hibernate.internal.SessionImpl;
import org.hibernate.jpa.AvailableSettings;
import org.junit.Test;
import org.hibernate.jpa.test.BaseEntityManagerFunctionalTestCase;
import org.hibernate.testing.TestForIssue;
import org.hibernate.testing.jta.TestingJtaBootstrap;
import org.hibernate.testing.jta.TestingJtaPlatformImpl;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
/**
* Largely a copy of {@link org.hibernate.test.jpa.txn.TransactionJoiningTest}
@ -138,4 +142,49 @@ public class TransactionJoiningTest extends BaseEntityManagerFunctionalTestCase
);
TestingJtaPlatformImpl.INSTANCE.getTransactionManager().commit();
}
/**
* In certain JTA environments (JBossTM, etc.), a background thread (reaper)
* can rollback a transaction if it times out. These timeouts are rare and
* typically come from server failures. However, we need to handle the
* multi-threaded nature of the transaction afterCompletion action.
* Emulate a timeout with a simple afterCompletion call in a thread.
* See HHH-7910
*/
@Test
@TestForIssue(jiraKey="HHH-7910")
public void testMultiThreadTransactionTimeout() throws Exception {
TestingJtaPlatformImpl.INSTANCE.getTransactionManager().begin();
EntityManager em = entityManagerFactory().createEntityManager();
final SessionImpl sImpl = em.unwrap( SessionImpl.class );
final CountDownLatch latch = new CountDownLatch(1);
Thread thread = new Thread() {
public void run() {
sImpl.getTransactionCoordinator().getSynchronizationCallbackCoordinator().afterCompletion( Status.STATUS_ROLLEDBACK );
latch.countDown();
}
};
thread.start();
latch.await();
em.persist( new Book( "The Book of Foo", 1 ) );
// Ensure that the session was cleared by the background thread.
assertEquals( "The background thread did not clear the session as expected!",
0, em.createQuery( "from Book" ).getResultList().size() );
TestingJtaPlatformImpl.INSTANCE.getTransactionManager().commit();
em.close();
}
@Override
public Class[] getAnnotatedClasses() {
return new Class[] {
Book.class
};
}
}