HHH-7821 - Clears the batch collection fetching queue on flush.

Provides a safe guard against NPE in BatchFetchQueue.getCollectionBatch() by checking that CollectionEntry.getLoadedKey() is not null before adding it to the batch.

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

@@ -85,6 +85,8 @@ public class BatchFetchQueue {
 
 	/**
 	 * Clears all entries from this fetch queue.
+	 * <p/>
+	 * Called after flushing or clearing the session.
 	 */
 	public void clear() {
 		batchLoadableEntityKeys.clear();
@@ -126,16 +128,6 @@ public class BatchFetchQueue {
 		subselectsByEntityKey.remove( key );
 	}
 
-	/**
-	 * Clears all pending subselect fetches from the queue.
-	 * <p/>
-	 * Called after flushing.
-	 */
-	public void clearSubselects() {
-		subselectsByEntityKey.clear();
-	}
-
-
 	// entity batch support ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 	/**
@@ -289,6 +281,15 @@ public class BatchFetchQueue {
 			for ( Map.Entry<CollectionEntry, PersistentCollection> me : map.entrySet() ) {
 				final CollectionEntry ce = me.getKey();
 				final PersistentCollection collection = me.getValue();
+				
+				if ( ce.getLoadedKey() == null ) {
+					// the loadedKey of the collectionEntry might be null as it might have been reset to null
+					// (see for example Collections.processDereferencedCollection()
+					// and CollectionEntry.afterAction())
+					// though we clear the queue on flush, it seems like a good idea to guard
+					// against potentially null loadedKeys (which leads to various NPEs as demonstrated in HHH-7821).
+					continue;
+				}
 
 				if ( collection.wasInitialized() ) {
 					// should never happen

hibernate-core/src/main/java/org/hibernate/event/internal/AbstractFlushingEventListener.java

@@ -349,8 +349,10 @@ public abstract class AbstractFlushingEventListener implements Serializable {
 
 		final PersistenceContext persistenceContext = session.getPersistenceContext();
 		persistenceContext.getCollectionsByKey().clear();
-		persistenceContext.getBatchFetchQueue()
+		
-				.clearSubselects(); //the database has changed now, so the subselect results need to be invalidated
+		// the database has changed now, so the subselect results need to be invalidated
+		// the batch fetching queues should also be cleared - especially the collection batch fetching one
+		persistenceContext.getBatchFetchQueue().clear();
 
 		for ( Map.Entry<PersistentCollection, CollectionEntry> me : IdentityMap.concurrentEntries( persistenceContext.getCollectionEntries() ) ) {
 			CollectionEntry collectionEntry = me.getValue();