HHH-11830 - Shared Session memory leak via TransactionObserver reference

hibernate-core/src/main/java/org/hibernate/internal/AbstractSharedSessionContract.java

@@ -207,6 +207,9 @@ public abstract class AbstractSharedSessionContract implements SharedSessionCont
 	protected void addSharedSessionTransactionObserver(TransactionCoordinator transactionCoordinator) {
 	}
 
+	protected void removeSharedSessionTransactionObserver(TransactionCoordinator transactionCoordinator) {
+	}
+
 	@Override
 	public boolean shouldAutoJoinTransaction() {
 		return autoJoinTransactions;
@@ -299,6 +302,10 @@ public abstract class AbstractSharedSessionContract implements SharedSessionCont
 			currentHibernateTransaction.invalidate();
 		}
 
+		if ( transactionCoordinator != null ) {
+			removeSharedSessionTransactionObserver( transactionCoordinator );
+		}
+
 		try {
 			if ( shouldCloseJdbcCoordinatorOnClose( isTransactionCoordinatorShared ) ) {
 				jdbcCoordinator.close();

hibernate-core/src/main/java/org/hibernate/internal/SessionImpl.java

@@ -246,6 +246,8 @@ public final class SessionImpl
 
 	private transient boolean discardOnClose;
 
+	private transient TransactionObserver transactionObserver;
+
 	public SessionImpl(SessionFactoryImpl factory, SessionCreationOptions options) {
 		super( factory, options );
 
@@ -2645,8 +2647,7 @@ public final class SessionImpl
 
 	@Override
 	protected void addSharedSessionTransactionObserver(TransactionCoordinator transactionCoordinator) {
-		transactionCoordinator.addObserver(
+		this.transactionObserver = new TransactionObserver() {
-				new TransactionObserver() {
 			@Override
 			public void afterBegin() {
 			}
@@ -2672,8 +2673,13 @@ public final class SessionImpl
 					managedClose();
 				}
 			}
+		};
+		transactionCoordinator.addObserver(transactionObserver);
 	}
-		);
+
+	@Override
+	protected void removeSharedSessionTransactionObserver(TransactionCoordinator transactionCoordinator) {
+		transactionCoordinator.removeObserver( transactionObserver );
 	}
 
 	private class IdentifierLoadAccessImpl<T> implements IdentifierLoadAccess<T> {

hibernate-core/src/main/java/org/hibernate/resource/transaction/backend/jdbc/internal/JdbcResourceLocalTransactionCoordinatorImpl.java

@@ -59,12 +59,22 @@ public class JdbcResourceLocalTransactionCoordinatorImpl implements TransactionC
 			TransactionCoordinatorBuilder transactionCoordinatorBuilder,
 			TransactionCoordinatorOwner owner,
 			JdbcResourceTransactionAccess jdbcResourceTransactionAccess) {
-		this.observers = new ArrayList<TransactionObserver>();
+		this.observers = new ArrayList<>();
 		this.transactionCoordinatorBuilder = transactionCoordinatorBuilder;
 		this.jdbcResourceTransactionAccess = jdbcResourceTransactionAccess;
 		this.transactionCoordinatorOwner = owner;
 	}
 
+	/**
+	 * Needed because while iterating the observers list and executing the before/update callbacks,
+	 * some observers might get removed from the list.
+	 *
+	 * @return TransactionObserver
+	 */
+	private Iterable<TransactionObserver> observers() {
+		return new ArrayList<>( observers );
+	}
+
 	@Override
 	public TransactionDriver getTransactionDriverControl() {
 		// Again, this PhysicalTransactionDelegate will act as the bridge from the local transaction back into the
@@ -134,7 +144,7 @@ public class JdbcResourceLocalTransactionCoordinatorImpl implements TransactionC
 			transactionCoordinatorOwner.setTransactionTimeOut( this.timeOut );
 		}
 		transactionCoordinatorOwner.afterTransactionBegin();
-		for ( TransactionObserver observer : observers ) {
+		for ( TransactionObserver observer : observers() ) {
 			observer.afterBegin();
 		}
 		log.trace( "ResourceLocalTransactionCoordinatorImpl#afterBeginCallback" );
@@ -145,7 +155,7 @@ public class JdbcResourceLocalTransactionCoordinatorImpl implements TransactionC
 		try {
 			transactionCoordinatorOwner.beforeTransactionCompletion();
 			synchronizationRegistry.notifySynchronizationsBeforeTransactionCompletion();
-			for ( TransactionObserver observer : observers ) {
+			for ( TransactionObserver observer : observers() ) {
 				observer.beforeCompletion();
 			}
 		}
@@ -164,11 +174,12 @@ public class JdbcResourceLocalTransactionCoordinatorImpl implements TransactionC
 		synchronizationRegistry.notifySynchronizationsAfterTransactionCompletion( statusToSend );
 
 		transactionCoordinatorOwner.afterTransactionCompletion( successful, false );
-		for ( TransactionObserver observer : observers ) {
+		for ( TransactionObserver observer : observers() ) {
 			observer.afterCompletion( successful, false );
 		}
 	}
 
+	@Override
 	public void addObserver(TransactionObserver observer) {
 		observers.add( observer );
 	}

hibernate-core/src/main/java/org/hibernate/resource/transaction/backend/jta/internal/JtaTransactionCoordinatorImpl.java

@@ -118,6 +118,16 @@ public class JtaTransactionCoordinatorImpl implements TransactionCoordinator, Sy
 
 	}
 
+	/**
+	 * Needed because while iterating the observers list and executing the before/update callbacks,
+	 * some observers might get removed from the list.
+	 *
+	 * @return TransactionObserver
+	 */
+	private Iterable<TransactionObserver> observers() {
+		return new ArrayList<>( observers );
+	}
+
 	public SynchronizationCallbackCoordinator getSynchronizationCallbackCoordinator() {
 		if ( callbackCoordinator == null ) {
 			callbackCoordinator = performJtaThreadTracking
@@ -329,7 +339,7 @@ public class JtaTransactionCoordinatorImpl implements TransactionCoordinator, Sy
 		}
 		finally {
 			synchronizationRegistry.notifySynchronizationsBeforeTransactionCompletion();
-			for ( TransactionObserver observer : observers ) {
+			for ( TransactionObserver observer : observers() ) {
 				observer.beforeCompletion();
 			}
 		}
@@ -348,7 +358,7 @@ public class JtaTransactionCoordinatorImpl implements TransactionCoordinator, Sy
 
 		transactionCoordinatorOwner.afterTransactionCompletion( successful, delayed );
 
-		for ( TransactionObserver observer : observers ) {
+		for ( TransactionObserver observer : observers() ) {
 			observer.afterCompletion( successful, delayed );
 		}
 

hibernate-core/src/test/java/org/hibernate/sharedSession/SessionWithSharedConnectionTest.java

@@ -6,6 +6,7 @@
  */
 package org.hibernate.sharedSession;
 
+import org.hibernate.FlushMode;
 import org.hibernate.IrrelevantEntity;
 import org.hibernate.Session;
 import org.hibernate.engine.spi.SessionImplementor;
@@ -15,10 +16,14 @@ import org.hibernate.event.spi.PostInsertEvent;
 import org.hibernate.event.spi.PostInsertEventListener;
 import org.hibernate.persister.entity.EntityPersister;
 
+import org.hibernate.resource.jdbc.spi.JdbcSessionOwner;
 import org.hibernate.testing.TestForIssue;
 import org.hibernate.testing.junit4.BaseCoreFunctionalTestCase;
 import org.junit.Test;
 
+import java.lang.reflect.Field;
+import java.util.Collection;
+
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -243,6 +248,50 @@ public class SessionWithSharedConnectionTest extends BaseCoreFunctionalTestCase
 		session.getTransaction().commit();
 	}
 
+	@Test
+	@TestForIssue(jiraKey = "HHH-11830")
+	public void testSharedSessionTransactionObserver() throws Exception {
+		Session session = openSession();
+
+		session.getTransaction().begin();
+
+		Field field = null;
+		Class<?> clazz = ((JdbcSessionOwner) session).getTransactionCoordinator().getClass();
+		while (clazz != null) {
+			try {
+				field = clazz.getDeclaredField("observers");
+				field.setAccessible(true);
+				break;
+			} catch (NoSuchFieldException e) {
+				clazz = clazz.getSuperclass();
+			} catch (Exception e) {
+				throw new IllegalStateException(e);
+			}
+		}
+		assertNotNull("Observers field was not found", field);
+
+		assertEquals(0, ((Collection) field.get(((SessionImplementor) session).getTransactionCoordinator())).size());
+
+		//open secondary sessions with managed options and immediately close
+		Session secondarySession;
+		for (int i = 0; i < 10; i++){
+			secondarySession = session.sessionWithOptions()
+					.connection()
+					.flushMode( FlushMode.COMMIT )
+					.autoClose( true )
+					.openSession();
+
+			//when the shared session is opened it should register an observer
+			assertEquals(1, ((Collection) field.get(((JdbcSessionOwner) session).getTransactionCoordinator())).size());
+
+			//observer should be released
+			secondarySession.close();
+
+			assertEquals(0, ((Collection) field.get(((JdbcSessionOwner) session).getTransactionCoordinator())).size());
+		}
+	}
+
+
 	@Override
 	protected Class<?>[] getAnnotatedClasses() {
 		return new Class[] { IrrelevantEntity.class };