HHH-18546 Clean up any hanging BulkOperationCleanupAction after-txn callbacks on Session close

# Conflicts:
#	hibernate-core/src/main/java/org/hibernate/internal/SessionImpl.java

Author: dreab8

hibernate-core/src/main/java/org/hibernate/action/internal/BulkOperationCleanupAction.java

@@ -220,7 +220,30 @@ public BeforeTransactionCompletionProcess getBeforeTransactionCompletionProcess(
 
 	@Override
 	public AfterTransactionCompletionProcess getAfterTransactionCompletionProcess() {
-		return (success, session) -> {
+		return new BulkOperationCleanUpAfterTransactionCompletionProcess(
+				entityCleanups,
+				collectionCleanups,
+				naturalIdCleanups
+		);
+	}
+
+	public static final class BulkOperationCleanUpAfterTransactionCompletionProcess
+			implements AfterTransactionCompletionProcess {
+		private final Set<EntityCleanup> entityCleanups;
+		private final Set<CollectionCleanup> collectionCleanups;
+		private final Set<NaturalIdCleanup> naturalIdCleanups;
+
+		public BulkOperationCleanUpAfterTransactionCompletionProcess(
+				Set<EntityCleanup> entityCleanups,
+				Set<CollectionCleanup> collectionCleanups,
+				Set<NaturalIdCleanup> naturalIdCleanups) {
+			this.entityCleanups = entityCleanups;
+			this.collectionCleanups = collectionCleanups;
+			this.naturalIdCleanups = naturalIdCleanups;
+		}
+
+		@Override
+		public void doAfterTransactionCompletion(boolean success, SharedSessionContractImplementor session) {
 			for ( EntityCleanup cleanup : entityCleanups ) {
 				cleanup.release();
 			}
@@ -235,7 +258,7 @@ public AfterTransactionCompletionProcess getAfterTransactionCompletionProcess()
 				cleanup.release();
 			}
 			collectionCleanups.clear();
-		};
+		}
 	}
 
 	@Override

hibernate-core/src/main/java/org/hibernate/engine/spi/ActionQueue.java

@@ -21,10 +21,12 @@
 
 import org.hibernate.AssertionFailure;
 import org.hibernate.HibernateException;
+import org.hibernate.Internal;
 import org.hibernate.PropertyValueException;
 import org.hibernate.TransientObjectException;
 import org.hibernate.action.internal.AbstractEntityInsertAction;
 import org.hibernate.action.internal.BulkOperationCleanupAction;
+import org.hibernate.action.internal.BulkOperationCleanupAction.BulkOperationCleanUpAfterTransactionCompletionProcess;
 import org.hibernate.action.internal.CollectionRecreateAction;
 import org.hibernate.action.internal.CollectionRemoveAction;
 import org.hibernate.action.internal.CollectionUpdateAction;
@@ -544,6 +546,16 @@ public void afterTransactionCompletion(boolean success) {
 		}
 	}
 
+	@Internal
+	public void executePendingBulkOperationCleanUpActions() {
+		if ( !isTransactionCoordinatorShared ) {
+			// Execute completion actions only in transaction owner (aka parent session).
+			if ( afterTransactionProcesses != null ) {
+				afterTransactionProcesses.executePendingBulkOperationCleanUpActions();
+			}
+		}
+	}
+
 	/**
 	 * Execute any registered {@link BeforeTransactionCompletionProcess}
 	 */
@@ -1071,6 +1083,39 @@ public void afterTransactionCompletion(boolean success) {
 			}
 			querySpacesToInvalidate.clear();
 		}
+
+		public void executePendingBulkOperationCleanUpActions() {
+			AfterTransactionCompletionProcess process;
+			boolean hasPendingBulkOperationCleanUpActions = false;
+			while ( ( process = processes.poll() ) != null ) {
+				if ( process instanceof BulkOperationCleanUpAfterTransactionCompletionProcess ) {
+					try {
+						hasPendingBulkOperationCleanUpActions = true;
+						process.doAfterTransactionCompletion( true, session );
+					}
+					catch (CacheException ce) {
+						LOG.unableToReleaseCacheLock( ce );
+						// continue loop
+					}
+					catch (Exception e) {
+						throw new HibernateException(
+								"Unable to perform afterTransactionCompletion callback: " + e.getMessage(),
+								e
+						);
+					}
+				}
+			}
+
+			if ( hasPendingBulkOperationCleanUpActions ) {
+				if ( session.getFactory().getSessionFactoryOptions().isQueryCacheEnabled() ) {
+					session.getFactory().getCache().getTimestampsCache().invalidate(
+							querySpacesToInvalidate.toArray( new String[0] ),
+							session
+					);
+				}
+				querySpacesToInvalidate.clear();
+			}
+		}
 	}
 
 	/**

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

@@ -449,6 +449,10 @@ public void closeWithoutOpenChecks() {
 			}
 		}
 		finally {
+			if ( actionQueue.hasAfterTransactionActions() ) {
+				log.warn( "Closing session with unprocessed clean up bulk operations, forcing their execution" );
+				actionQueue.executePendingBulkOperationCleanUpActions();
+			}
 			final StatisticsImplementor statistics = getSessionFactory().getStatistics();
 			if ( statistics.isStatisticsEnabled() ) {
 				statistics.closeSession();