#1519 - Memory leak when using @Cache(enableQueryCache=true) (#1523)

* #1519 - Memory leak when using @Cache(enableQueryCache=true)

* #1519 - Memory leak when using @Cache(enableQueryCache=true)

Remove unused TransactionEventBeans

src/main/java/io/ebeaninternal/api/TransactionEvent.java

@@ -9,6 +9,7 @@ import io.ebeaninternal.server.transaction.TransactionManager;
 import io.ebeanservice.docstore.api.DocStoreUpdates;
 
 import java.io.Serializable;
+import java.util.ArrayList;
 import java.util.List;
 
 /**
@@ -29,16 +30,21 @@ public class TransactionEvent implements Serializable {
    */
   private final transient boolean local;
 
+  private final long startMillis;
+
   private TransactionEventTable eventTables;
 
-  private transient TransactionEventBeans eventBeans;
+  private transient List<PersistRequestBean<?>> listenerNotify;
 
   private transient DeleteByIdMap deleteByIdMap;
 
+  private transient CacheChangeSet changeSet;
+
   /**
    * Create the TransactionEvent, one per Transaction.
    */
-  public TransactionEvent() {
+  public TransactionEvent(long startMillis) {
+    this.startMillis = startMillis;
     this.local = true;
   }
 
@@ -71,8 +77,8 @@ public class TransactionEvent implements Serializable {
   /**
    * Return the list of PersistRequestBean's for this transaction.
    */
-  public List<PersistRequestBean<?>> getPersistRequestBeans() {
-    return (eventBeans == null) ? null : eventBeans.getRequests();
+  public List<PersistRequestBean<?>> getListenerNotify() {
+    return (listenerNotify == null) ? null : listenerNotify;
   }
 
   public TransactionEventTable getEventTables() {
@@ -94,17 +100,13 @@ public class TransactionEvent implements Serializable {
   }
 
   /**
-   * Add a inserted updated or deleted bean to the event.
+   * Add post commit listeners. Watch this for large transactions.
    */
-  public void add(PersistRequestBean<?> request) {
-
-    if (request.isNotify()) {
-      // either a BeanListener or Cache is interested
-      if (eventBeans == null) {
-        eventBeans = new TransactionEventBeans();
-      }
-      eventBeans.add(request);
+  public void addListenerNotify(PersistRequestBean<?> request) {
+    if (listenerNotify == null) {
+      listenerNotify = new ArrayList<>();
     }
+    listenerNotify.add(request);
   }
 
   /**
@@ -112,11 +114,13 @@ public class TransactionEvent implements Serializable {
    */
   public CacheChangeSet buildCacheChanges(TransactionManager manager) {
 
-    if (eventBeans == null && deleteByIdMap == null && eventTables == null) {
+    if (changeSet == null && deleteByIdMap == null && eventTables == null) {
       return null;
     }
 
-    CacheChangeSet changeSet = new CacheChangeSet(manager.clockNowMillis());
+    if (changeSet == null) {
+      changeSet = new CacheChangeSet(manager.clockNowMillis());
+    }
     if (eventTables != null && !eventTables.isEmpty()) {
       // notify cache with table based changes
       BeanDescriptorManager dm = manager.getBeanDescriptorManager();
@@ -124,9 +128,6 @@ public class TransactionEvent implements Serializable {
         dm.cacheNotify(tableIUD, changeSet);
       }
     }
-    if (eventBeans != null) {
-      eventBeans.notifyCache(changeSet);
-    }
     if (deleteByIdMap != null) {
       deleteByIdMap.notifyCache(changeSet);
     }
@@ -137,12 +138,25 @@ public class TransactionEvent implements Serializable {
    * Add any relevant PersistRequestBean's to DocStoreUpdates for later processing.
    */
   public void addDocStoreUpdates(DocStoreUpdates docStoreUpdates) {
-
-    List<PersistRequestBean<?>> persistRequestBeans = getPersistRequestBeans();
-    if (persistRequestBeans != null) {
-      for (PersistRequestBean<?> persistRequestBean : persistRequestBeans) {
+    List<PersistRequestBean<?>> requests = getListenerNotify();
+    if (requests != null) {
+      for (PersistRequestBean<?> persistRequestBean : requests) {
         persistRequestBean.addDocStoreUpdates(docStoreUpdates);
       }
     }
   }
+
+  /**
+   * Return the CacheChangeSet that we add cache notification messages to.
+   *
+   * We want to add to this change set as we process requests allowing the
+   * PersistRequestBean to be garbage collected for large transactions.
+   */
+  public CacheChangeSet obtainCacheChangeSet() {
+    if (changeSet == null) {
+      changeSet = new CacheChangeSet(startMillis);
+    }
+    return changeSet;
+  }
+
 }

src/main/java/io/ebeaninternal/api/TransactionEventBeans.java

@@ -1,44 +0,0 @@
-package io.ebeaninternal.api;
-
-import io.ebeaninternal.server.cache.CacheChangeSet;
-import io.ebeaninternal.server.core.PersistRequestBean;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * Lists of inserted updated and deleted beans that have a BeanPersistListener.
- * <p>
- * These beans will be sent to the appropriate BeanListeners after a successful
- * commit of the transaction.
- * </p>
- */
-public class TransactionEventBeans {
-
-  final ArrayList<PersistRequestBean<?>> requests = new ArrayList<>();
-
-  /**
-   * Return the list of PersistRequests that BeanListeners are interested in.
-   */
-  public List<PersistRequestBean<?>> getRequests() {
-    return requests;
-  }
-
-  /**
-   * Add a bean for BeanListener notification.
-   */
-  public void add(PersistRequestBean<?> request) {
-
-    requests.add(request);
-  }
-
-  /**
-   * Collect the cache changes.
-   */
-  public void notifyCache(CacheChangeSet changeSet) {
-    for (PersistRequestBean<?> request : requests) {
-      request.notifyCache(changeSet);
-    }
-  }
-
-}

src/main/java/io/ebeaninternal/server/core/PersistRequestBean.java

@@ -173,6 +173,16 @@ public final class PersistRequestBean<T> extends PersistRequest implements BeanP
    */
   private int pendingPostUpdateNotify;
 
+  /**
+   * Set to true when post execute has occured (so includes batch flush).
+   */
+  private boolean postExecute;
+
+  /**
+   * Set to true after many properties have been persisted (so includes element collections).
+   */
+  private boolean complete;
+
   public PersistRequestBean(SpiEbeanServer server, T bean, Object parentBean, BeanManager<T> mgr, SpiTransaction t,
                             PersistExecute persistExecute, PersistRequest.Type type, int flags) {
 
@@ -441,10 +451,10 @@ public final class PersistRequestBean<T> extends PersistRequest implements BeanP
   }
 
   /**
-   * Return true if this change should notify cache, listener or doc store.
+   * Return true if this change should notify persist listener or doc store (and keep the request).
    */
-  public boolean isNotify() {
-    return notifyCache || isNotifyPersistListener() || isDocStoreNotify();
+  private boolean isNotifyListeners() {
+    return isNotifyPersistListener() || isDocStoreNotify();
   }
 
   /**
@@ -918,7 +928,7 @@ public final class PersistRequestBean<T> extends PersistRequest implements BeanP
    */
   @Override
   public void postExecute() {
-
+    postExecute = true;
     if (controller != null) {
       controllerPost();
     }
@@ -939,7 +949,8 @@ public final class PersistRequestBean<T> extends PersistRequest implements BeanP
       postInsert();
     }
 
-    addEvent();
+    addPostCommitListeners();
+    notifyCacheOnPostExecute();
 
     if (isLogSummary()) {
       logSummary();
@@ -1017,14 +1028,12 @@ public final class PersistRequestBean<T> extends PersistRequest implements BeanP
   }
 
   /**
-   * Add the bean to the TransactionEvent. This will be used by TransactionManager to sync Cache,
-   * Cluster and text indexes.
+   * Add the request to TransactionEvent if there are post commit listeners.
    */
-  private void addEvent() {
-
+  private void addPostCommitListeners() {
     TransactionEvent event = transaction.getEvent();
-    if (event != null) {
-      event.add(this);
+    if (event != null && isNotifyListeners()) {
+      event.addListenerNotify(this);
     }
   }
 
@@ -1087,23 +1096,62 @@ public final class PersistRequestBean<T> extends PersistRequest implements BeanP
   }
 
   /**
-   * Check if any of its many properties where cascade saved and hence we need to update related
-   * many property caches.
+   * Completed insert or delete request. Do cache notify in non-batched case.
    */
-  public void checkUpdatedManysOnly() {
+  public void complete() {
+    notifyCacheOnComplete();
+  }
+
+  /**
+   * Completed update request handling cases for element collection and where ONLY
+   * many properties were updated.
+   */
+  public void completeUpdate() {
     if (!dirty && updatedManys != null) {
       // set the flag and register for post commit processing if there
       // is caching or registered listeners
       if (idValue == null) {
         this.idValue = beanDescriptor.getId(entityBean);
       }
+      // not dirty so postExecute() will never be called
+      // set true to trigger cache notify if needed
+      postExecute = true;
       updatedManysOnly = true;
       setNotifyCache();
-      addEvent();
+      addPostCommitListeners();
     }
+    notifyCacheOnComplete();
     postUpdateNotify();
   }
 
+  /**
+   * Notify cache when using batched persist.
+   */
+  private void notifyCacheOnPostExecute() {
+    postExecute = true;
+    if (notifyCache && complete) {
+      // add cache notification (on batch persist)
+      TransactionEvent event = transaction.getEvent();
+      if (event != null) {
+        notifyCache(event.obtainCacheChangeSet());
+      }
+    }
+  }
+
+  /**
+   * Notify cache when not use batched persist.
+   */
+  private void notifyCacheOnComplete() {
+    complete = true;
+    if (notifyCache && postExecute) {
+      // add cache notification (on non-batch persist)
+      TransactionEvent event = transaction.getEvent();
+      if (event != null) {
+        notifyCache(event.obtainCacheChangeSet());
+      }
+    }
+  }
+
   /**
    * For requests that update document store add this event to either the list
    * of queue events or list of update events.
@@ -1390,7 +1438,7 @@ public final class PersistRequestBean<T> extends PersistRequest implements BeanP
   public void setImportedOrphanForRemoval(BeanPropertyAssocOne<?> prop) {
     Object orphan = getOrigValue(prop);
     if (orphan instanceof EntityBean) {
-      orphanBean = (EntityBean)orphan;
+      orphanBean = (EntityBean) orphan;
     }
   }
 

src/main/java/io/ebeaninternal/server/deploy/BeanCollectionUtil.java

@@ -22,7 +22,7 @@ public class BeanCollectionUtil {
   }
 
   /**
-   * Return the details of the collection or map taking care to avoid
+   * Return the details (map entry set) of the collection or map taking care to avoid
    * unnecessary fetching of the data.
    */
   public static Collection<?> getActualEntries(Object o) {
@@ -48,4 +48,32 @@ public class BeanCollectionUtil {
     }
     throw new PersistenceException("expecting a Map or Collection but got [" + o.getClass().getName() + "]");
   }
+
+  /**
+   * Return the details (map values) of the collection or map taking care to avoid
+   * unnecessary fetching of the data.
+   */
+  public static Collection<?> getActualDetails(Object o) {
+    if (o == null) {
+      return null;
+    }
+    if (o instanceof BeanCollection<?>) {
+      BeanCollection<?> bc = (BeanCollection<?>) o;
+      if (!bc.isPopulated()) {
+        return null;
+      }
+      // For maps this is a collection of Map.Entry, otherwise it
+      // returns a collection of beans
+      return bc.getActualDetails();
+    }
+
+    if (o instanceof Map<?, ?>) {
+      // yes, we want the entrySet (to set the keys)
+      return ((Map<?, ?>) o).values();
+
+    } else if (o instanceof Collection<?>) {
+      return ((Collection<?>) o);
+    }
+    throw new PersistenceException("expecting a Map or Collection but got [" + o.getClass().getName() + "]");
+  }
 }

src/main/java/io/ebeaninternal/server/deploy/BeanDescriptor.java

@@ -7,7 +7,6 @@ import io.ebean.SqlUpdate;
 import io.ebean.Transaction;
 import io.ebean.ValuePair;
 import io.ebean.annotation.DocStoreMode;
-import io.ebean.annotation.PartitionMode;
 import io.ebean.bean.BeanCollection;
 import io.ebean.bean.EntityBean;
 import io.ebean.bean.EntityBeanIntercept;

src/main/java/io/ebeaninternal/server/deploy/BeanDescriptorCacheHelp.java

@@ -348,13 +348,12 @@ final class BeanDescriptorCacheHelp<T> {
 
   private CachedManyIds createManyIds(BeanPropertyAssocMany<?> many, Object details) {
 
-    BeanDescriptor<?> targetDescriptor = many.getTargetDescriptor();
-
-    Collection<?> actualDetails = BeanCollectionUtil.getActualEntries(details);
+    Collection<?> actualDetails = BeanCollectionUtil.getActualDetails(details);
     if (actualDetails == null) {
       return null;
     }
 
+    BeanDescriptor<?> targetDescriptor = many.getTargetDescriptor();
     List<Object> idList = new ArrayList<>(actualDetails.size());
     for (Object bean : actualDetails) {
       idList.add(targetDescriptor.getId((EntityBean) bean));
@@ -794,7 +793,6 @@ final class BeanDescriptorCacheHelp<T> {
         // query caching only
         return;
       }
-
       List<BeanPropertyAssocMany<?>> manyCollections = updateRequest.getUpdatedManyCollections();
       if (manyCollections != null) {
         for (BeanPropertyAssocMany<?> many : manyCollections) {

src/main/java/io/ebeaninternal/server/persist/DefaultPersister.java

@@ -412,7 +412,7 @@ public final class DefaultPersister implements Persister {
         if (req.isPersistCascade()) {
           saveAssocMany(req);
         }
-        req.checkUpdatedManysOnly();
+        req.completeUpdate();
       } else {
         update(req);
       }
@@ -472,7 +472,7 @@ public final class DefaultPersister implements Persister {
         request.flagUpdate();
         saveAssocMany(request);
       }
-      request.checkUpdatedManysOnly();
+      request.completeUpdate();
     } else {
       if (request.isInsert()) {
         insert(request);
@@ -506,6 +506,8 @@ public final class DefaultPersister implements Persister {
         // save any associated List held beans
         saveAssocMany(request);
       }
+      request.complete();
+
     } finally {
       request.unRegisterBean();
     }
@@ -539,7 +541,7 @@ public final class DefaultPersister implements Persister {
         saveAssocMany(request);
       }
 
-      request.checkUpdatedManysOnly();
+      request.completeUpdate();
 
     } finally {
       request.unRegisterBean();
@@ -883,6 +885,7 @@ public final class DefaultPersister implements Persister {
         unloadedForeignKeys.deleteCascade();
       }
     }
+    request.complete();
 
     // return true if using JDBC batch (as we can't tell until the batch is flushed)
     return count;

src/main/java/io/ebeaninternal/server/transaction/JdbcTransaction.java

@@ -14,7 +14,6 @@ import io.ebeaninternal.api.SpiTransaction;
 import io.ebeaninternal.api.TransactionEvent;
 import io.ebeaninternal.api.TxnProfileEventCodes;
 import io.ebeaninternal.server.core.PersistDeferredRelationship;
-import io.ebeaninternal.server.core.PersistRequest;
 import io.ebeaninternal.server.core.PersistRequestBean;
 import io.ebeaninternal.server.lib.util.Str;
 import io.ebeaninternal.server.persist.BatchControl;
@@ -855,7 +854,7 @@ public class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes {
   public TransactionEvent getEvent() {
     queryOnly = false;
     if (event == null) {
-      event = new TransactionEvent();
+      event = new TransactionEvent(startMillis);
     }
     return event;
   }
@@ -1052,7 +1051,7 @@ public class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes {
       // the event has been sent to the transaction manager
       // for postCommit processing (l2 cache updates etc)
       // start a new transaction event
-      event = new TransactionEvent();
+      event = new TransactionEvent(startMillis);
 
     } catch (Exception e) {
       doRollback(e);

src/main/java/io/ebeaninternal/server/transaction/PostCommitProcessing.java

@@ -33,9 +33,7 @@ final class PostCommitProcessing {
 
   private final TransactionManager manager;
 
-  private final List<PersistRequestBean<?>> persistBeanRequests;
-
-  private final BeanPersistIdMap beanPersistIdMap;
+  private final List<PersistRequestBean<?>> listenerNotify;
 
   private final RemoteTransactionEvent remoteTransactionEvent;
 
@@ -57,8 +55,7 @@ final class PostCommitProcessing {
     this.txnDocStoreBatchSize = 0;
     this.event = event;
     this.deleteByIdMap = event.getDeleteByIdMap();
-    this.persistBeanRequests = event.getPersistRequestBeans();
-    this.beanPersistIdMap = createBeanPersistIdMap();
+    this.listenerNotify = event.getListenerNotify();
     this.remoteTransactionEvent = createRemoteTransactionEvent();
   }
 
@@ -74,8 +71,7 @@ final class PostCommitProcessing {
     this.txnDocStoreBatchSize = transaction.getDocStoreBatchSize();
     this.event = transaction.getEvent();
     this.deleteByIdMap = event.getDeleteByIdMap();
-    this.persistBeanRequests = event.getPersistRequestBeans();
-    this.beanPersistIdMap = createBeanPersistIdMap();
+    this.listenerNotify = event.getListenerNotify();
     this.remoteTransactionEvent = createRemoteTransactionEvent();
   }
 
@@ -159,9 +155,9 @@ final class PostCommitProcessing {
   }
 
   private void localPersistListenersNotify() {
-    if (persistBeanRequests != null) {
-      for (PersistRequestBean<?> persistBeanRequest : persistBeanRequests) {
-        persistBeanRequest.notifyLocalPersistListener();
+    if (listenerNotify != null) {
+      for (PersistRequestBean<?> request : listenerNotify) {
+        request.notifyLocalPersistListener();
       }
     }
     TransactionEventTable eventTables = event.getEventTables();
@@ -175,13 +171,13 @@ final class PostCommitProcessing {
 
   private BeanPersistIdMap createBeanPersistIdMap() {
 
-    if (persistBeanRequests == null) {
+    if (listenerNotify == null) {
       return null;
     }
 
     BeanPersistIdMap m = new BeanPersistIdMap();
-    for (PersistRequestBean<?> persistBeanRequest : persistBeanRequests) {
-      persistBeanRequest.addToPersistMap(m);
+    for (PersistRequestBean<?> request : listenerNotify) {
+      request.addToPersistMap(m);
     }
     return m;
   }
@@ -193,6 +189,7 @@ final class PostCommitProcessing {
     }
 
     RemoteTransactionEvent remoteTransactionEvent = new RemoteTransactionEvent(serverName);
+    BeanPersistIdMap beanPersistIdMap = createBeanPersistIdMap();
     if (beanPersistIdMap != null) {
       for (BeanPersistIds beanPersist : beanPersistIdMap.values()) {
         remoteTransactionEvent.addBeanPersistIds(beanPersist);

src/main/java/io/ebeaninternal/server/transaction/TransactionManager.java

@@ -452,7 +452,7 @@ public class TransactionManager implements SpiTransactionManager {
 
   private void externalModificationEvent(TransactionEventTable tableEvents) {
 
-    TransactionEvent event = new TransactionEvent();
+    TransactionEvent event = new TransactionEvent(clockNowMillis());
     event.add(tableEvents);
 
     PostCommitProcessing postCommit = new PostCommitProcessing(clusterManager, this, event);