Merge branch 'feature/query-cache-1427'

src/main/java/io/ebean/cache/QueryCacheEntry.java

@@ -0,0 +1,54 @@
+package io.ebean.cache;
+
+import java.util.Set;
+
+/**
+ * For query cache entries we additionally hold the dependent tables and timestamp for the query result.
+ * <p>
+ * We use the dependent tables and timestamp to validate that tables the query joins to have not been
+ * modified since the query cache entry was cached. If any dependent tables have since been modified
+ * the query cache entry is treated as invalid.
+ * </p>
+ */
+public class QueryCacheEntry {
+
+  private final Object value;
+
+  private final Set<String> dependentTables;
+
+  private final long timestamp;
+
+  /**
+   * Create with dependent tables and timestamp.
+   *
+   * @param value           The query result being cached
+   * @param dependentTables The extra tables the query is dependent on (joins to)
+   * @param timestamp       The timestamp that the query uses to check for modifications
+   */
+  public QueryCacheEntry(Object value, Set<String> dependentTables, long timestamp) {
+    this.value = value;
+    this.dependentTables = dependentTables;
+    this.timestamp = timestamp;
+  }
+
+  /**
+   * Return the actual query result.
+   */
+  public Object getValue() {
+    return value;
+  }
+
+  /**
+   * Return the tables the query result is dependent on.
+   */
+  public Set<String> getDependentTables() {
+    return dependentTables;
+  }
+
+  /**
+   * Return the timestamp used to check for modifications on the dependent tables.
+   */
+  public long getTimestamp() {
+    return timestamp;
+  }
+}

src/main/java/io/ebean/cache/QueryCacheEntryValidate.java

@@ -0,0 +1,12 @@
+package io.ebean.cache;
+
+/**
+ * Used to validate that a query cache entry is still valid based on dependent tables.
+ */
+public interface QueryCacheEntryValidate {
+
+  /**
+   * Return true if the entry is still valid based on dependent tables.
+   */
+  boolean isValid(QueryCacheEntry queryCacheEntry);
+}

src/main/java/io/ebean/cache/ServerCacheConfig.java

@@ -0,0 +1,65 @@
+package io.ebean.cache;
+
+import io.ebean.config.CurrentTenantProvider;
+
+/**
+ * Configuration used to create ServerCache instances.
+ */
+public class ServerCacheConfig {
+
+  private final ServerCacheType type;
+  private final String cacheKey;
+  private final ServerCacheOptions cacheOptions;
+  private final CurrentTenantProvider tenantProvider;
+  private final QueryCacheEntryValidate queryCacheEntryValidate;
+
+  public ServerCacheConfig(ServerCacheType type, String cacheKey, ServerCacheOptions cacheOptions, CurrentTenantProvider tenantProvider, QueryCacheEntryValidate queryCacheEntryValidate) {
+    this.type = type;
+    this.cacheKey = cacheKey;
+    this.cacheOptions = cacheOptions;
+    this.tenantProvider = tenantProvider;
+    this.queryCacheEntryValidate = queryCacheEntryValidate;
+  }
+
+  /**
+   * Return the cache type.
+   */
+  public ServerCacheType getType() {
+    return type;
+  }
+
+  /**
+   * Return the name of the cache.
+   */
+  public String getCacheKey() {
+    return cacheKey;
+  }
+
+  /**
+   * Return the tuning options.
+   */
+  public ServerCacheOptions getCacheOptions() {
+    return cacheOptions;
+  }
+
+  /**
+   * Return the current tenant provider.
+   */
+  public CurrentTenantProvider getTenantProvider() {
+    return tenantProvider;
+  }
+
+  /**
+   * Return the service that provides validation for query cache entries.
+   */
+  public QueryCacheEntryValidate getQueryCacheEntryValidate() {
+    return queryCacheEntryValidate;
+  }
+
+  /**
+   * Return true if the cache is a query cache.
+   */
+  public boolean isQueryCache() {
+    return type == ServerCacheType.QUERY;
+  }
+}

src/main/java/io/ebean/cache/ServerCacheFactory.java

@@ -1,7 +1,5 @@
 package io.ebean.cache;
 
-import io.ebean.config.CurrentTenantProvider;
-
 /**
  * Defines method for constructing caches for beans and queries.
  */
@@ -10,6 +8,15 @@ public interface ServerCacheFactory {
   /**
    * Create the cache for the given type with options.
    */
-  ServerCache createCache(ServerCacheType type, String cacheKey, CurrentTenantProvider tenantProvider, ServerCacheOptions cacheOptions);
+  ServerCache createCache(ServerCacheConfig config);
 
+  /**
+   * Return a ServerCacheNotify that we will send ServerCacheNotification events to.
+   * <p>
+   * This is used if a ServerCacheNotifyPlugin is not supplied.
+   * </p>
+   *
+   * @param listener The listener that should be used to process the notification events.
+   */
+  ServerCacheNotify createCacheNotify(ServerCacheNotify listener);
 }

src/main/java/io/ebean/cache/ServerCacheNotification.java

@@ -0,0 +1,29 @@
+package io.ebean.cache;
+
+import java.util.Set;
+
+/**
+ * Notification event that dependent tables have been modified.
+ * <p>
+ * This is sent to other interested servers (in the cluster).
+ * </p>
+ */
+public class ServerCacheNotification {
+
+  private final long modifyTimestamp;
+
+  private final Set<String> dependentTables;
+
+  public ServerCacheNotification(long modifyTimestamp, Set<String> dependentTables) {
+    this.modifyTimestamp = modifyTimestamp;
+    this.dependentTables = dependentTables;
+  }
+
+  public long getModifyTimestamp() {
+    return modifyTimestamp;
+  }
+
+  public Set<String> getDependentTables() {
+    return dependentTables;
+  }
+}

src/main/java/io/ebean/cache/ServerCacheNotify.java

@@ -0,0 +1,12 @@
+package io.ebean.cache;
+
+/**
+ * Interface for both listening to notification changes and sending them to other members of the cluster.
+ */
+public interface ServerCacheNotify {
+
+  /**
+   * Notify other server cache members of the table modifications or process the notifications.
+   */
+  void notify(ServerCacheNotification notification);
+}

src/main/java/io/ebean/cache/ServerCacheNotifyPlugin.java

@@ -0,0 +1,17 @@
+package io.ebean.cache;
+
+import io.ebean.config.ServerConfig;
+
+/**
+ * Plugin that provides a ServerCacheNotify implementation.
+ * <p>
+ * Is supplied this will be used to send the ServerCacheNotification event to other cluster members.
+ * </p>
+ */
+public interface ServerCacheNotifyPlugin {
+
+  /**
+   * Create a ServerCacheNotify implementation given the server configuration.
+   */
+  ServerCacheNotify create(ServerConfig serverConfig);
+}

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

@@ -101,6 +101,11 @@ public interface SpiTransaction extends Transaction {
    */
   String getId();
 
+  /**
+   * Return the start timestamp for the transaction (JVM side).
+   */
+  long getStartMillis();
+
   /**
    * Return true if this transaction has updateAllLoadedProperties set.
    * If null is returned the server default is used (set on ServerConfig).

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

@@ -30,6 +30,11 @@ public abstract class SpiTransactionProxy implements SpiTransaction {
     return transaction.translate(message, cause);
   }
 
+  @Override
+  public long getStartMillis() {
+    return transaction.getStartMillis();
+  }
+
   @Override
   public void setLabel(String label) {
     transaction.setLabel(label);

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

@@ -27,6 +27,8 @@ public class TransactionEvent implements Serializable {
    */
   private final transient boolean local;
 
+  private final long modificationTimestamp;
+
   private TransactionEventTable eventTables;
 
   private transient TransactionEventBeans eventBeans;
@@ -36,8 +38,9 @@ public class TransactionEvent implements Serializable {
   /**
    * Create the TransactionEvent, one per Transaction.
    */
-  public TransactionEvent() {
+  public TransactionEvent(long modificationTimestamp) {
     this.local = true;
+    this.modificationTimestamp = modificationTimestamp;
   }
 
   public void addDeleteById(BeanDescriptor<?> desc, Object id) {
@@ -108,8 +111,8 @@ public class TransactionEvent implements Serializable {
   /**
    * Build and return the cache changeSet.
    */
-  public CacheChangeSet buildCacheChanges(boolean viewInvalidation) {
-    CacheChangeSet changeSet = new CacheChangeSet(viewInvalidation);
+  public CacheChangeSet buildCacheChanges() {
+    CacheChangeSet changeSet = new CacheChangeSet(modificationTimestamp);
     if (eventBeans != null) {
       eventBeans.notifyCache(changeSet);
     }

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

@@ -1,8 +1,8 @@
 package io.ebeaninternal.api;
 
 import io.ebean.event.BulkTableEvent;
-import io.ebeaninternal.server.cluster.BinaryMessage;
-import io.ebeaninternal.server.cluster.BinaryMessageList;
+import io.ebeaninternal.server.cluster.binarymessage.BinaryMessage;
+import io.ebeaninternal.server.cluster.binarymessage.BinaryMessageList;
 
 import java.io.DataInput;
 import java.io.DataOutputStream;
@@ -102,7 +102,7 @@ public final class TransactionEventTable implements Serializable {
       os.writeBoolean(insert);
       os.writeBoolean(update);
       os.writeBoolean(delete);
-
+      os.close();
       msgList.add(msg);
     }
 

src/main/java/io/ebeaninternal/server/cache/CacheChangeSet.java

@@ -17,22 +17,26 @@ public class CacheChangeSet {
 
   private final List<CacheChange> entries = new ArrayList<>();
 
+  private final Set<String> touchedTables = new HashSet<>();
+
   private final Set<BeanDescriptor<?>> queryCaches = new HashSet<>();
 
   private final Map<ManyKey, ManyChange> manyChangeMap = new HashMap<>();
 
-  /**
-   * Set of "base tables" modified used to invalidate entities based on views.
-   */
-  private final Set<String> viewInvalidation = new HashSet<>();
-
-  private final boolean viewEntityInvalidation;
+  private final long modificationTimestamp;
 
   /**
    * Construct specifying if we also need to process invalidation for entities based on views.
    */
-  public CacheChangeSet(boolean viewEntityInvalidation) {
-    this.viewEntityInvalidation = viewEntityInvalidation;
+  public CacheChangeSet(long modificationTimestamp) {
+    this.modificationTimestamp = modificationTimestamp;
+  }
+
+  /**
+   * Return the touched tables.
+   */
+  public Set<String> touchedTables() {
+    return touchedTables;
   }
 
   /**
@@ -40,7 +44,7 @@ public class CacheChangeSet {
    * <p>
    * Return the set of table changes to process invalidation for entities based on views.
    */
-  public Set<String> apply() {
+  public void apply() {
     for (BeanDescriptor<?> entry : queryCaches) {
       entry.clearQueryCache();
     }
@@ -50,7 +54,13 @@ public class CacheChangeSet {
     for (CacheChange entry : manyChangeMap.values()) {
       entry.apply();
     }
-    return viewInvalidation;
+  }
+
+  /**
+   * Add an entry to clear a query cache.
+   */
+  public void addInvalidate(BeanDescriptor<?> descriptor) {
+    touchedTables.add(descriptor.getBaseTable());
   }
 
   /**
@@ -85,29 +95,23 @@ public class CacheChangeSet {
    * On bean insert register table for view based entity invalidation.
    */
   public void addBeanInsert(String baseTable) {
-    if (viewEntityInvalidation) {
-      viewInvalidation.add(baseTable);
-    }
+    touchedTables.add(baseTable);
   }
 
   /**
    * Remove a bean from the cache.
    */
   public <T> void addBeanRemove(BeanDescriptor<T> desc, Object id) {
+    touchedTables.add(desc.getBaseTable());
     entries.add(new CacheChangeBeanRemove(desc, id));
-    if (viewEntityInvalidation) {
-      viewInvalidation.add(desc.getBaseTable());
-    }
   }
 
   /**
    * Update a bean entry.
    */
   public <T> void addBeanUpdate(BeanDescriptor<T> desc, Object id, Map<String, Object> changes, boolean updateNaturalKey, long version) {
+    touchedTables.add(desc.getBaseTable());
     entries.add(new CacheChangeBeanUpdate(desc, id, changes, updateNaturalKey, version));
-    if (viewEntityInvalidation) {
-      viewInvalidation.add(desc.getBaseTable());
-    }
   }
 
   /**
@@ -125,6 +129,13 @@ public class CacheChangeSet {
     return manyChangeMap.computeIfAbsent(key, ManyChange::new);
   }
 
+  /**
+   * Return the modification timestamp for these changes.
+   */
+  public long modificationTimestamp() {
+    return modificationTimestamp;
+  }
+
   /**
    * Changes for a specific many property.
    */

src/main/java/io/ebeaninternal/server/cache/CacheManagerOptions.java

@@ -1,5 +1,6 @@
 package io.ebeaninternal.server.cache;
 
+import io.ebean.cache.QueryCacheEntryValidate;
 import io.ebean.cache.ServerCacheFactory;
 import io.ebean.cache.ServerCacheOptions;
 import io.ebean.config.CurrentTenantProvider;
@@ -19,6 +20,8 @@ public class CacheManagerOptions {
 
   private CurrentTenantProvider currentTenantProvider;
 
+  private QueryCacheEntryValidate queryCacheEntryValidate;
+
   private ServerCacheFactory cacheFactory = new DefaultServerCacheFactory();
   private ServerCacheOptions beanDefault = new ServerCacheOptions();
   private ServerCacheOptions queryDefault = new ServerCacheOptions();
@@ -45,8 +48,9 @@ public class CacheManagerOptions {
     return this;
   }
 
-  public CacheManagerOptions with(ServerCacheFactory cacheFactory) {
+  public CacheManagerOptions with(ServerCacheFactory cacheFactory, QueryCacheEntryValidate queryCacheEntryValidate) {
     this.cacheFactory = cacheFactory;
+    this.queryCacheEntryValidate = queryCacheEntryValidate;
     return this;
   }
 
@@ -82,4 +86,8 @@ public class CacheManagerOptions {
   public ClusterManager getClusterManager() {
     return clusterManager;
   }
+
+  public QueryCacheEntryValidate getQueryCacheEntryValidate() {
+    return queryCacheEntryValidate;
+  }
 }

src/main/java/io/ebeaninternal/server/cache/DefaultCacheHolder.java

@@ -2,7 +2,9 @@ package io.ebeaninternal.server.cache;
 
 import io.ebean.annotation.CacheBeanTuning;
 import io.ebean.annotation.CacheQueryTuning;
+import io.ebean.cache.QueryCacheEntryValidate;
 import io.ebean.cache.ServerCache;
+import io.ebean.cache.ServerCacheConfig;
 import io.ebean.cache.ServerCacheFactory;
 import io.ebean.cache.ServerCacheOptions;
 import io.ebean.cache.ServerCacheType;
@@ -33,22 +35,14 @@ class DefaultCacheHolder {
 
   private final CurrentTenantProvider tenantProvider;
 
-  DefaultCacheHolder(CacheManagerOptions builder) {
-    this(builder.getCacheFactory(), builder.getBeanDefault(), builder.getQueryDefault(), builder.getCurrentTenantProvider());
-  }
+  private final QueryCacheEntryValidate queryCacheEntryValidate;
 
-  /**
-   * Create with a cache factory and default cache options.
-   *
-   * @param cacheFactory the factory for creating the cache
-   * @param beanDefault  the default options for tuning bean caches
-   * @param queryDefault the default options for tuning query caches
-   */
-  DefaultCacheHolder(ServerCacheFactory cacheFactory, ServerCacheOptions beanDefault, ServerCacheOptions queryDefault, CurrentTenantProvider tenantProvider) {
-    this.cacheFactory = cacheFactory;
-    this.beanDefault = beanDefault;
-    this.queryDefault = queryDefault;
-    this.tenantProvider = tenantProvider;
+  DefaultCacheHolder(CacheManagerOptions builder) {
+    this.cacheFactory = builder.getCacheFactory();
+    this.beanDefault = builder.getBeanDefault();
+    this.queryDefault = builder.getQueryDefault();
+    this.tenantProvider = builder.getCurrentTenantProvider();
+    this.queryCacheEntryValidate = builder.getQueryCacheEntryValidate();
   }
 
   ServerCache getCache(Class<?> beanType, String cacheKey, ServerCacheType type) {
@@ -76,7 +70,7 @@ class DefaultCacheHolder {
         collectIdCaches.computeIfAbsent(beanType.getName(), s -> new ConcurrentSkipListSet<>()).add(key);
       }
     }
-    return cacheFactory.createCache(type, key, tenantProvider, options);
+    return cacheFactory.createCache(new ServerCacheConfig(type, key, options, tenantProvider, queryCacheEntryValidate));
   }
 
   void clearAll() {

src/main/java/io/ebeaninternal/server/cache/DefaultServerCache.java

@@ -2,10 +2,8 @@ package io.ebeaninternal.server.cache;
 
 import io.ebean.BackgroundExecutor;
 import io.ebean.cache.ServerCache;
-import io.ebean.cache.ServerCacheOptions;
 import io.ebean.cache.ServerCacheStatistics;
 import io.ebean.cache.TenantAwareKey;
-import io.ebean.config.CurrentTenantProvider;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -15,7 +13,6 @@ import java.util.Comparator;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.LongAdder;
 
@@ -65,47 +62,14 @@ public class DefaultServerCache implements ServerCache {
 
   protected TenantAwareKey tenantAwareKey;
 
-  /**
-   * Construct using a ConcurrentHashMap and cache options.
-   */
-  public DefaultServerCache(String name, CurrentTenantProvider tenantProvider, ServerCacheOptions options) {
-    this(name, new ConcurrentHashMap<>(), tenantProvider, options);
-  }
-
-  /**
-   * Construct passing in name, map and base eviction controls as ServerCacheOptions.
-   */
-  public DefaultServerCache(String name, Map<Object, CacheEntry> map, CurrentTenantProvider tenantProvider, ServerCacheOptions options) {
-    this(name, map, tenantProvider, options.getMaxSize(), options.getMaxIdleSecs(), options.getMaxSecsToLive(), options.getTrimFrequency());
-  }
-
-  /**
-   * Construct passing in name, map and base eviction controls.
-   */
-  public DefaultServerCache(String name, Map<Object, CacheEntry> map, CurrentTenantProvider tenantProvider, int maxSize, int maxIdleSecs, int maxSecsToLive, int trimFrequency) {
-    this.name = name;
-    this.map = map;
-    this.maxSize = maxSize;
-    this.tenantAwareKey = new TenantAwareKey(tenantProvider);
-    this.maxIdleSecs = maxIdleSecs;
-    this.maxSecsToLive = maxSecsToLive;
-    this.trimFrequency = determineTrim(maxIdleSecs, maxSecsToLive, trimFrequency);
-  }
-
-  /**
-   * Determine a good trimFrequency as half of maxIdleSecs (or maxSecsToLive).
-   */
-  int determineTrim(int maxIdleSecs, int maxSecsToLive, int trimFrequency) {
-    if (trimFrequency > 0) {
-      return trimFrequency;
-    }
-    if (maxIdleSecs > 0) {
-      return maxIdleSecs / 2 - 1;
-    }
-    if (maxSecsToLive > 0) {
-      return maxSecsToLive / 2 - 1;
-    }
-    return 0;
+  public DefaultServerCache(DefaultServerCacheConfig config) {
+    this.name = config.getName();
+    this.map = config.getMap();
+    this.maxSize = config.getMaxSize();
+    this.tenantAwareKey = new TenantAwareKey(config.getTenantProvider());
+    this.maxIdleSecs = config.getMaxIdleSecs();
+    this.maxSecsToLive = config.getMaxSecsToLive();
+    this.trimFrequency = config.determineTrimFrequency();
   }
 
   public void periodicTrim(BackgroundExecutor executor) {
@@ -193,7 +157,7 @@ public class DefaultServerCache implements ServerCache {
   /**
    * Return the tenant aware key.
    */
-  private Object key(Object id) {
+  protected Object key(Object id) {
     return tenantAwareKey.key(id);
   }
 
@@ -203,7 +167,7 @@ public class DefaultServerCache implements ServerCache {
   @Override
   public Object get(Object id) {
 
-    CacheEntry entry = map.get(key(id));
+    CacheEntry entry = getCacheEntry(id);
     if (entry == null) {
       missCount.increment();
       return null;
@@ -212,10 +176,24 @@ public class DefaultServerCache implements ServerCache {
       // Important that hitCount.increment() MUST be low latency under concurrent
       // use hence must use LongAdder or better here
       hitCount.increment();
-      return entry.getValue();
+      return unwrapEntry(entry);
     }
   }
 
+  /**
+   * Unwrap the cache entry - override for query cache to unwrap to the query result.
+   */
+  protected Object unwrapEntry(CacheEntry entry) {
+    return entry.getValue();
+  }
+
+  /**
+   * Get the cache entry - override for query cache to validate dependent tables.
+   */
+  protected CacheEntry getCacheEntry(Object id) {
+    return map.get(key(id));
+  }
+
   @Override
   public void putAll(Map<Object, Object> keyValues) {
     keyValues.forEach(this::put);

src/main/java/io/ebeaninternal/server/cache/DefaultServerCacheConfig.java

@@ -0,0 +1,80 @@
+package io.ebeaninternal.server.cache;
+
+import io.ebean.cache.QueryCacheEntryValidate;
+import io.ebean.cache.ServerCacheConfig;
+import io.ebean.cache.ServerCacheOptions;
+import io.ebean.config.CurrentTenantProvider;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class DefaultServerCacheConfig {
+
+  private final ServerCacheConfig config;
+
+  private int maxSize;
+  private int maxIdleSecs;
+  private int maxSecsToLive;
+  private int trimFrequency;
+
+  private Map<Object, DefaultServerCache.CacheEntry> map;
+
+  public DefaultServerCacheConfig(ServerCacheConfig config) {
+    this(config,  new ConcurrentHashMap<>());
+  }
+
+  public DefaultServerCacheConfig(ServerCacheConfig config, Map<Object, DefaultServerCache.CacheEntry> map) {
+    this.config = config;
+    this.map = map;
+
+    ServerCacheOptions options = config.getCacheOptions();
+    this.maxIdleSecs = options.getMaxIdleSecs();
+    this.maxSecsToLive = options.getMaxSecsToLive();
+    this.trimFrequency = options.getTrimFrequency();
+    this.maxSize = options.getMaxSize();
+  }
+
+  public CurrentTenantProvider getTenantProvider() {
+    return config.getTenantProvider();
+  }
+
+  public QueryCacheEntryValidate getQueryCacheEntryValidate() {
+    return config.getQueryCacheEntryValidate();
+  }
+
+  public String getName() {
+    return config.getCacheKey();
+  }
+
+  public Map<Object, DefaultServerCache.CacheEntry> getMap() {
+    return map;
+  }
+
+  public int getMaxSize() {
+    return maxSize;
+  }
+
+  public int getMaxIdleSecs() {
+    return maxIdleSecs;
+  }
+
+  public int getMaxSecsToLive() {
+    return maxSecsToLive;
+  }
+
+  /**
+   * Determine a good trimFrequency as half of maxIdleSecs (or maxSecsToLive).
+   */
+  public int determineTrimFrequency() {
+    if (trimFrequency > 0) {
+      return trimFrequency;
+    }
+    if (maxIdleSecs > 0) {
+      return maxIdleSecs / 2 - 1;
+    }
+    if (maxSecsToLive > 0) {
+      return maxSecsToLive / 2 - 1;
+    }
+    return 0;
+  }
+}

src/main/java/io/ebeaninternal/server/cache/DefaultServerCacheFactory.java

@@ -2,10 +2,10 @@ package io.ebeaninternal.server.cache;
 
 import io.ebean.BackgroundExecutor;
 import io.ebean.cache.ServerCache;
+import io.ebean.cache.ServerCacheConfig;
 import io.ebean.cache.ServerCacheFactory;
-import io.ebean.cache.ServerCacheOptions;
-import io.ebean.cache.ServerCacheType;
-import io.ebean.config.CurrentTenantProvider;
+import io.ebean.cache.ServerCacheNotification;
+import io.ebean.cache.ServerCacheNotify;
 
 
 /**
@@ -18,25 +18,43 @@ class DefaultServerCacheFactory implements ServerCacheFactory {
   /**
    * Construct when l2 cache is disabled.
    */
-  public DefaultServerCacheFactory() {
+  DefaultServerCacheFactory() {
     this.executor = null;
   }
 
   /**
    * Construct with executor service.
    */
-  public DefaultServerCacheFactory(BackgroundExecutor executor) {
+  DefaultServerCacheFactory(BackgroundExecutor executor) {
     this.executor = executor;
   }
 
   @Override
-  public ServerCache createCache(ServerCacheType type, String cacheKey, CurrentTenantProvider tenantProvider, ServerCacheOptions cacheOptions) {
+  public ServerCache createCache(ServerCacheConfig config) {
 
-    DefaultServerCache cache = new DefaultServerCache(cacheKey, tenantProvider, cacheOptions);
+    DefaultServerCache cache;
+    if (config.isQueryCache()) {
+      // use a server cache aware of extra validation and QueryCacheEntry
+      cache = new DefaultServerQueryCache(new DefaultServerCacheConfig(config));
+    } else {
+      cache = new DefaultServerCache(new DefaultServerCacheConfig(config));
+    }
     if (executor != null) {
       cache.periodicTrim(executor);
     }
     return cache;
   }
 
+  @Override
+  public ServerCacheNotify createCacheNotify(ServerCacheNotify listener) {
+    return new NoopServerCacheNotify();
+  }
+
+  private static class NoopServerCacheNotify implements ServerCacheNotify {
+
+    @Override
+    public void notify(ServerCacheNotification notification) {
+      // do nothing
+    }
+  }
 }

src/main/java/io/ebeaninternal/server/cache/DefaultServerQueryCache.java

@@ -0,0 +1,41 @@
+package io.ebeaninternal.server.cache;
+
+import io.ebean.cache.QueryCacheEntry;
+import io.ebean.cache.QueryCacheEntryValidate;
+
+/**
+ * Server cache for query caching.
+ * <p>
+ * Entries in this cache contain QueryCacheEntry and we need to additionally
+ * validate the entries when hit for changes to dependent tables.
+ * </p>
+ */
+public class DefaultServerQueryCache extends DefaultServerCache {
+
+  private final QueryCacheEntryValidate queryCacheEntryValidate;
+
+  public DefaultServerQueryCache(DefaultServerCacheConfig config) {
+    super(config);
+    this.queryCacheEntryValidate = config.getQueryCacheEntryValidate();
+  }
+
+  protected Object unwrapEntry(CacheEntry entry) {
+    return ((QueryCacheEntry) entry.getValue()).getValue();
+  }
+
+  @Override
+  protected CacheEntry getCacheEntry(Object id) {
+    Object key = key(id);
+    CacheEntry entry = map.get(key);
+    if (entry == null) {
+      return null;
+    }
+    QueryCacheEntry value = (QueryCacheEntry) entry.getValue();
+    if (!queryCacheEntryValidate.isValid(value)) {
+      map.remove(key);
+      removeCount.increment();
+      return null;
+    }
+    return entry;
+  }
+}

src/main/java/io/ebeaninternal/server/cache/RemoteCacheEvent.java

@@ -1,7 +1,7 @@
 package io.ebeaninternal.server.cache;
 
-import io.ebeaninternal.server.cluster.BinaryMessage;
-import io.ebeaninternal.server.cluster.BinaryMessageList;
+import io.ebeaninternal.server.cluster.binarymessage.BinaryMessage;
+import io.ebeaninternal.server.cluster.binarymessage.BinaryMessageList;
 
 import java.io.DataInput;
 import java.io.DataOutputStream;

src/main/java/io/ebeaninternal/server/cluster/ClusterManager.java

@@ -13,7 +13,7 @@ import java.util.concurrent.ConcurrentHashMap;
 /**
  * Manages the cluster service.
  */
-public class ClusterManager {
+public class ClusterManager implements MessageServerProvider {
 
   private static final Logger clusterLogger = LoggerFactory.getLogger("io.ebean.Cluster");
 

src/main/java/io/ebeaninternal/server/cluster/MessageServerProvider.java

@@ -0,0 +1,14 @@
+package io.ebeaninternal.server.cluster;
+
+import io.ebean.EbeanServer;
+
+/**
+ * Returns EbeanServer instances for remote message reading.
+ */
+public interface MessageServerProvider {
+
+  /**
+   * Return the EbeanServer instance by name.
+   */
+  EbeanServer getServer(String name);
+}

src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryDataReader.java

@@ -0,0 +1,75 @@
+package io.ebeaninternal.server.cluster.binarymessage;
+
+import io.ebeaninternal.api.SpiEbeanServer;
+import io.ebeaninternal.api.TransactionEventTable;
+import io.ebeaninternal.server.cache.RemoteCacheEvent;
+import io.ebeaninternal.server.cluster.MessageServerProvider;
+import io.ebeaninternal.server.transaction.BeanPersistIds;
+import io.ebeaninternal.server.transaction.RemoteTableMod;
+import io.ebeaninternal.server.transaction.RemoteTransactionEvent;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+
+/**
+ * Reads the binary message returning RemoteTransactionEvent.
+ */
+class BinaryDataReader {
+
+  private final MessageServerProvider serverProvider;
+  private final DataInputStream dataInput;
+
+  private SpiEbeanServer server;
+  private RemoteTransactionEvent event;
+
+  BinaryDataReader(MessageServerProvider serverProvider, byte[] data) {
+    this.serverProvider = serverProvider;
+    this.dataInput = new DataInputStream(new ByteArrayInputStream(data));
+  }
+
+  /**
+   * Read the binary message returning a RemoteTransactionEvent.
+   */
+  RemoteTransactionEvent read() throws IOException {
+
+    String serverName = dataInput.readUTF();
+
+    this.server = (SpiEbeanServer) serverProvider.getServer(serverName);
+    if (server == null) {
+      throw new IllegalStateException("EbeanServer not found for name [" + serverName + "]");
+    }
+    this.event = new RemoteTransactionEvent(server);
+    boolean more = dataInput.readBoolean();
+    while (more) {
+      readMessage();
+      more = dataInput.readBoolean();
+    }
+    return event;
+  }
+
+  private void readMessage() throws IOException {
+
+    int msgType = dataInput.readInt();
+    switch (msgType) {
+      case BinaryMessage.TYPE_BEANIUD:
+        event.addBeanPersistIds(BeanPersistIds.readBinaryMessage(server, dataInput));
+        break;
+
+      case BinaryMessage.TYPE_TABLEIUD:
+        event.addTableIUD(TransactionEventTable.TableIUD.readBinaryMessage(dataInput));
+        break;
+
+      case BinaryMessage.TYPE_CACHE:
+        event.addRemoteCacheEvent(RemoteCacheEvent.readBinaryMessage(dataInput));
+        break;
+
+      case BinaryMessage.TYPE_TABLEMOD:
+        event.addRemoteTableMod(RemoteTableMod.readBinaryMessage(dataInput));
+        break;
+
+      default:
+        throw new RuntimeException("Invalid Transaction msgType " + msgType);
+    }
+  }
+}

src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryDataWriter.java

@@ -0,0 +1,49 @@
+package io.ebeaninternal.server.cluster.binarymessage;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+/**
+ * Writes BinaryMessageList to DataHolder raw byte[].
+ */
+class BinaryDataWriter {
+
+  private final BinaryMessageList messageList;
+
+  private final ByteArrayOutputStream buffer;
+
+  private final DataOutputStream dataOut;
+
+  private final String serverName;
+
+  BinaryDataWriter(String serverName, BinaryMessageList messageList) {
+    this.serverName = serverName;
+    this.messageList = messageList;
+    this.buffer = new ByteArrayOutputStream(256);
+    this.dataOut = new DataOutputStream(buffer);
+  }
+
+  /**
+   * Write the message as raw byte[].
+   */
+  byte[] write() throws IOException {
+
+    //TODO: Add Protocol version and "podId" to support pubsub cluster
+    //dataOut.writeInt(MsgKeys.MESSAGE_KEY);
+    dataOut.writeUTF(serverName);
+    for (BinaryMessage msg : messageList.getList()) {
+      write(msg);
+    }
+    dataOut.writeBoolean(false);
+    dataOut.flush();
+    return buffer.toByteArray();
+  }
+
+  private void write(BinaryMessage msg) throws IOException {
+
+    dataOut.writeBoolean(true);
+    dataOut.write(msg.getByteArray());
+  }
+
+}

src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryMessage.java

@@ -1,4 +1,4 @@
-package io.ebeaninternal.server.cluster;
+package io.ebeaninternal.server.cluster.binarymessage;
 
 import java.io.ByteArrayOutputStream;
 import java.io.DataOutputStream;
@@ -24,6 +24,7 @@ public class BinaryMessage {
   public static final int TYPE_BEANIUD = 1;
   public static final int TYPE_TABLEIUD = 2;
   public static final int TYPE_CACHE = 3;
+  public static final int TYPE_TABLEMOD = 4;
 
   public static final int TYPE_MSGACK = 8;
   public static final int TYPE_MSGRESEND = 9;

src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryMessageList.java

@@ -1,4 +1,4 @@
-package io.ebeaninternal.server.cluster;
+package io.ebeaninternal.server.cluster.binarymessage;
 
 import java.util.ArrayList;
 import java.util.List;

src/main/java/io/ebeaninternal/server/cluster/binarymessage/ClusterMessage.java

@@ -0,0 +1,147 @@
+package io.ebeaninternal.server.cluster.binarymessage;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+/**
+ * The message broadcast around the cluster.
+ */
+public class ClusterMessage {
+
+  private static final int MAX_LENGTH = 10 * 1024 * 1024;
+
+  private final String registerIp;
+
+  private final String podName;
+
+  private final boolean register;
+
+  private final byte[] data;
+
+  /**
+   * Create a register message.
+   */
+  public static ClusterMessage register(String registerIp, boolean register, String podName) {
+    return new ClusterMessage(registerIp, register, podName);
+  }
+
+  /**
+   * Create a transaction message.
+   */
+  public static ClusterMessage transEvent(byte[] data) {
+    return new ClusterMessage(data);
+  }
+
+  /**
+   * Create for register online/offline message.
+   */
+  private ClusterMessage(String registerIp, boolean register, String podName) {
+    this.registerIp = registerIp;
+    this.register = register;
+    this.podName = podName;
+    this.data = null;
+  }
+
+  /**
+   * Create for a transaction message.
+   */
+  private ClusterMessage(byte[] data) {
+    this.data = data;
+    this.registerIp = null;
+    this.podName = null;
+    this.register = false;
+  }
+
+  public String toString() {
+    StringBuilder sb = new StringBuilder();
+    if (registerIp != null) {
+      sb.append("register ");
+      sb.append(register);
+      sb.append(" ");
+      sb.append(registerIp);
+    } else {
+      sb.append("[data]");
+    }
+    return sb.toString();
+  }
+
+  /**
+   * Return true if this is a register event as opposed to a transaction message.
+   */
+  public boolean isRegisterEvent() {
+    return registerIp != null;
+  }
+
+  /**
+   * Return the register host for online/offline message.
+   */
+  public String getRegisterIp() {
+    return registerIp;
+  }
+
+  public String getPodName() {
+    return podName;
+  }
+
+  /**
+   * Return true if register is true for a online/offline message.
+   */
+  public boolean isRegister() {
+    return register;
+  }
+
+  /**
+   * Return the raw message data.
+   */
+  public byte[] getData() {
+    return data;
+  }
+
+  /**
+   * Write the message in binary form.
+   */
+  public void write(DataOutputStream dataOutput) throws IOException {
+
+    if (data != null) {
+      // write data message
+      dataOutput.writeInt(MsgKeys.DATA);
+      dataOutput.writeInt(data.length);
+      dataOutput.write(data);
+    } else {
+      // write header message
+      dataOutput.writeInt(MsgKeys.HEADER);
+      dataOutput.writeUTF(getRegisterIp());
+      dataOutput.writeBoolean(register);
+      dataOutput.writeUTF(getPodName());
+    }
+    dataOutput.flush();
+  }
+
+  /**
+   * Read the message from binary form.
+   */
+  public static ClusterMessage read(DataInputStream dataInput) throws IOException, InvalidMessageException {
+
+    int key = dataInput.readInt();
+    if (key == MsgKeys.DATA) {
+      int length = dataInput.readInt();
+      if (length > MAX_LENGTH) {
+        throw new IOException("Message data too large length:"+length);
+      }
+      byte[] data = new byte[length];
+      dataInput.readFully(data);
+      return new ClusterMessage(data);
+
+    } else if (key == MsgKeys.HEADER) {
+      String host = dataInput.readUTF();
+      boolean registered = dataInput.readBoolean();
+      String podName = dataInput.readUTF();
+      return new ClusterMessage(host, registered, podName);
+
+    } else {
+      throw new InvalidMessageException("Invalid message key:" + key);
+    }
+  }
+
+}

src/main/java/io/ebeaninternal/server/cluster/binarymessage/InvalidMessageException.java

@@ -0,0 +1,8 @@
+package io.ebeaninternal.server.cluster.binarymessage;
+
+public class InvalidMessageException extends Exception {
+
+	public InvalidMessageException(String message) {
+		super(message);
+	}
+}

src/main/java/io/ebeaninternal/server/cluster/binarymessage/MessageReadWrite.java

@@ -0,0 +1,37 @@
+package io.ebeaninternal.server.cluster.binarymessage;
+
+import io.ebeaninternal.server.cluster.MessageServerProvider;
+import io.ebeaninternal.server.transaction.RemoteTransactionEvent;
+
+import java.io.IOException;
+
+/**
+ * Mechanism to convert RemoteTransactionEvent to/from byte[] content.
+ */
+public class MessageReadWrite {
+
+  private final MessageServerProvider serverProvider;
+
+  public MessageReadWrite(MessageServerProvider serverProvider) {
+    this.serverProvider = serverProvider;
+  }
+
+  /**
+   * Convert the RemoteTransactionEvent to raw byte[] content.
+   */
+  public byte[] write(RemoteTransactionEvent transEvent) throws IOException {
+
+    BinaryMessageList messageList = new BinaryMessageList();
+    transEvent.writeBinaryMessage(messageList);
+
+    return new BinaryDataWriter(transEvent.getServerName(), messageList).write();
+  }
+
+  /**
+   * Convert the byte[] content to RemoteTransactionEvent.
+   */
+  public RemoteTransactionEvent read(byte[] data) throws IOException {
+
+    return new BinaryDataReader(serverProvider, data).read();
+  }
+}

src/main/java/io/ebeaninternal/server/cluster/binarymessage/MsgKeys.java

@@ -0,0 +1,19 @@
+package io.ebeaninternal.server.cluster.binarymessage;
+
+public interface MsgKeys {
+
+	/**
+	 * Used to identify client on connection initiation.
+	 */
+	int HELLO = 182;
+
+	/**
+	 * Used to confirm protocol on reading messages.
+	 */
+	int HEADER = 11;
+
+	/**
+	 * Used to confirm protocol on reading messages.
+	 */
+	int DATA = 12;
+}

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

@@ -12,6 +12,9 @@ public class CacheOptions {
    */
   public static final CacheOptions NO_CACHING = new CacheOptions();
 
+  public static final CacheOptions INVALIDATE_QUERY_CACHE = new CacheOptions(true);
+
+  private final boolean invalidateQueryCache;
   private final boolean enableBeanCache;
   private final boolean enableQueryCache;
   private final boolean readOnly;
@@ -21,6 +24,18 @@ public class CacheOptions {
    * Construct for no caching.
    */
   private CacheOptions() {
+    invalidateQueryCache = false;
+    enableBeanCache = false;
+    enableQueryCache = false;
+    readOnly = false;
+    naturalKey = null;
+  }
+
+  /**
+   * Construct for invalidateQueryCache.
+   */
+  private CacheOptions(boolean invalidateQueryCache) {
+    this.invalidateQueryCache = invalidateQueryCache;
     enableBeanCache = false;
     enableQueryCache = false;
     readOnly = false;
@@ -31,12 +46,21 @@ public class CacheOptions {
    * Construct with cache annotation.
    */
   public CacheOptions(Cache cache, String[] naturalKey) {
+    invalidateQueryCache = false;
     enableBeanCache = cache.enableBeanCache();
     enableQueryCache = cache.enableQueryCache();
     readOnly = cache.readOnly();
     this.naturalKey = naturalKey;
   }
 
+  /**
+   * Return true if this is InvalidateQueryCache. A Bean that itself isn't L2
+   * cached but invalidates query cache entries that join to it.
+   */
+  public boolean isInvalidateQueryCache() {
+    return invalidateQueryCache;
+  }
+
   /**
    * Return true if bean caching is enabled.
    */

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

@@ -0,0 +1,29 @@
+package io.ebeaninternal.server.core;
+
+import java.time.Clock;
+
+/**
+ * Wraps the Clock such that we can change the Clock for testing purposes.
+ */
+public class ClockService {
+
+  private Clock clock;
+
+  public ClockService(Clock clock) {
+    this.clock = clock;
+  }
+
+  /**
+   * Change the clock for testing purposes.
+   */
+  public void setClock(Clock clock) {
+    this.clock = clock;
+  }
+
+  /**
+   * Return the Clock current timestamp in millis.
+   */
+  public long nowMillis() {
+    return clock.millis();
+  }
+}

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

@@ -138,11 +138,10 @@ public class DefaultContainer implements SpiContainer {
 
       // executor and l2 caching service setup early (used during server construction)
       SpiBackgroundExecutor executor = createBackgroundExecutor(serverConfig);
-      SpiCacheManager cacheManager = getCacheManager(online, serverConfig, executor);
 
-      InternalConfiguration c = new InternalConfiguration(clusterManager, cacheManager, executor, serverConfig, bootupClasses);
+      InternalConfiguration c = new InternalConfiguration(online, clusterManager, executor, serverConfig, bootupClasses);
 
-      DefaultServer server = new DefaultServer(c, c.cache());
+      DefaultServer server = new DefaultServer(c, c.cacheManager());
 
       // generate and run DDL if required
       // if there are any other tasks requiring action in their plugins, do them as well
@@ -165,52 +164,6 @@ public class DefaultContainer implements SpiContainer {
     }
   }
 
-  /**
-   * Create and return the CacheManager.
-   */
-  private SpiCacheManager getCacheManager(boolean online, ServerConfig serverConfig, BackgroundExecutor executor) {
-
-    if (!online || serverConfig.isDisableL2Cache()) {
-      // use local only L2 cache implementation as placeholder
-      return new DefaultServerCacheManager();
-    }
-
-    // reasonable default settings are for a cache per bean type
-    ServerCacheOptions beanOptions = new ServerCacheOptions();
-    beanOptions.setMaxSize(serverConfig.getCacheMaxSize());
-    beanOptions.setMaxIdleSecs(serverConfig.getCacheMaxIdleTime());
-    beanOptions.setMaxSecsToLive(serverConfig.getCacheMaxTimeToLive());
-
-    // reasonable default settings for the query cache per bean type
-    ServerCacheOptions queryOptions = new ServerCacheOptions();
-    queryOptions.setMaxSize(serverConfig.getQueryCacheMaxSize());
-    queryOptions.setMaxIdleSecs(serverConfig.getQueryCacheMaxIdleTime());
-    queryOptions.setMaxSecsToLive(serverConfig.getQueryCacheMaxTimeToLive());
-
-    boolean localL2Caching = false;
-    ServerCachePlugin plugin = serverConfig.getServerCachePlugin();
-    if (plugin == null) {
-      ServiceLoader<ServerCachePlugin> cacheFactories = ServiceLoader.load(ServerCachePlugin.class);
-      Iterator<ServerCachePlugin> iterator = cacheFactories.iterator();
-      if (iterator.hasNext()) {
-        // use the cacheFactory (via classpath service loader)
-        plugin = iterator.next();
-        logger.debug("using ServerCacheFactory {}", plugin.getClass());
-      } else {
-        // use the built in default l2 caching which is local cache based
-        localL2Caching = true;
-        plugin = new DefaultServerCachePlugin();
-      }
-    }
-
-    ServerCacheFactory factory = plugin.create(serverConfig, executor);
-
-    CacheManagerOptions builder = new CacheManagerOptions(clusterManager, serverConfig, localL2Caching)
-      .with(beanOptions, queryOptions)
-      .with(factory);
-    return new DefaultServerCacheManager(builder);
-  }
-
   /**
    * Get the entities, scalarTypes, Listeners etc combining the class registered
    * ones with the already created instances.

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

@@ -154,7 +154,7 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
   /**
    * Clock to use for WhenModified and WhenCreated.
    */
-  private Clock clock;
+  private ClockService clockService;
 
   private final CallStackFactory callStackFactory;
 
@@ -283,7 +283,7 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
     this.beanLoader = new DefaultBeanLoader(this);
     this.jsonContext = config.createJsonContext(this);
     this.dataTimeZone = config.getDataTimeZone();
-    this.clock = config.getServerConfig().getClock();
+    this.clockService = config.getClockService();
 
     DocStoreIntegration docStoreComponents = config.createDocStoreIntegration(this);
     this.transactionManager = config.createTransactionManager(docStoreComponents.updateProcessor());
@@ -515,12 +515,12 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
 
   @Override
   public long clockNow() {
-    return clock.millis();
+    return clockService.nowMillis();
   }
 
   @Override
   public void setClock(Clock clock) {
-    this.clock = clock;
+    this.clockService.setClock(clock);
   }
 
   @Override

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

@@ -3,7 +3,12 @@ package io.ebeaninternal.server.core;
 import com.fasterxml.jackson.core.JsonFactory;
 import io.ebean.ExpressionFactory;
 import io.ebean.annotation.Platform;
+import io.ebean.cache.ServerCacheFactory;
 import io.ebean.cache.ServerCacheManager;
+import io.ebean.cache.ServerCacheNotify;
+import io.ebean.cache.ServerCacheNotifyPlugin;
+import io.ebean.cache.ServerCacheOptions;
+import io.ebean.cache.ServerCachePlugin;
 import io.ebean.config.ExternalTransactionManager;
 import io.ebean.config.ProfilingConfig;
 import io.ebean.config.ServerConfig;
@@ -27,7 +32,10 @@ import io.ebeaninternal.api.SpiProfileHandler;
 import io.ebeaninternal.dbmigration.DbOffline;
 import io.ebeaninternal.server.autotune.AutoTuneService;
 import io.ebeaninternal.server.autotune.service.AutoTuneServiceFactory;
+import io.ebeaninternal.server.cache.CacheManagerOptions;
 import io.ebeaninternal.server.cache.DefaultCacheAdapter;
+import io.ebeaninternal.server.cache.DefaultServerCacheManager;
+import io.ebeaninternal.server.cache.DefaultServerCachePlugin;
 import io.ebeaninternal.server.cache.SpiCacheManager;
 import io.ebeaninternal.server.changelog.DefaultChangeLogListener;
 import io.ebeaninternal.server.changelog.DefaultChangeLogPrepare;
@@ -69,6 +77,7 @@ import io.ebeaninternal.server.transaction.ExplicitTransactionManager;
 import io.ebeaninternal.server.transaction.ExternalTransactionScopeManager;
 import io.ebeaninternal.server.transaction.JtaTransactionManager;
 import io.ebeaninternal.server.transaction.NoopProfileHandler;
+import io.ebeaninternal.server.transaction.TableModState;
 import io.ebeaninternal.server.transaction.TransactionManager;
 import io.ebeaninternal.server.transaction.TransactionManagerOptions;
 import io.ebeaninternal.server.transaction.TransactionScopeManager;
@@ -84,6 +93,7 @@ import org.slf4j.LoggerFactory;
 
 import javax.sql.DataSource;
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.ServiceLoader;
@@ -96,6 +106,10 @@ public class InternalConfiguration {
 
   private static final Logger logger = LoggerFactory.getLogger(InternalConfiguration.class);
 
+  private final TableModState tableModState = new TableModState();
+
+  private final boolean online;
+
   private final ServerConfig serverConfig;
 
   private final BootupClasses bootupClasses;
@@ -108,6 +122,8 @@ public class InternalConfiguration {
 
   private final DtoBeanManager dtoBeanManager;
 
+  private final ClockService clockService;
+
   private final DataTimeZone dataTimeZone;
 
   private final Binder binder;
@@ -124,6 +140,12 @@ public class InternalConfiguration {
 
   private final SpiCacheManager cacheManager;
 
+  private final ServerCachePlugin serverCachePlugin;
+
+  private ServerCacheNotify cacheNotify;
+
+  private boolean localL2Caching;
+
   private final ExpressionFactory expressionFactory;
 
   private final SpiBackgroundExecutor backgroundExecutor;
@@ -141,17 +163,17 @@ public class InternalConfiguration {
 
   private final SpiLogManager logManager;
 
-  public InternalConfiguration(ClusterManager clusterManager,
-                               SpiCacheManager cacheManager, SpiBackgroundExecutor backgroundExecutor,
+  public InternalConfiguration(boolean online, ClusterManager clusterManager, SpiBackgroundExecutor backgroundExecutor,
                                ServerConfig serverConfig, BootupClasses bootupClasses) {
 
+    this.online = online;
     this.serverConfig = serverConfig;
+    this.clockService = new ClockService(serverConfig.getClock());
     this.logManager = initLogManager();
     this.docStoreFactory = initDocStoreFactory(serverConfig.service(DocStoreFactory.class));
     this.jsonFactory = serverConfig.getJsonFactory();
     this.clusterManager = clusterManager;
     this.backgroundExecutor = backgroundExecutor;
-    this.cacheManager = cacheManager;
     this.bootupClasses = bootupClasses;
 
     this.databasePlatform = serverConfig.getDatabasePlatform();
@@ -164,6 +186,9 @@ public class InternalConfiguration {
     this.deployCreateProperties = new DeployCreateProperties(typeManager);
     this.deployUtil = new DeployUtil(typeManager, serverConfig);
 
+    this.serverCachePlugin = initServerCachePlugin();
+    this.cacheManager = initCacheManager();
+
     InternalConfigXmlRead xmlRead = new InternalConfigXmlRead(serverConfig);
 
     this.dtoBeanManager = new DtoBeanManager(typeManager, xmlRead.readDtoMapping());
@@ -211,6 +236,10 @@ public class InternalConfiguration {
     return docStoreFactory;
   }
 
+  public ClockService getClockService() {
+    return clockService;
+  }
+
   /**
    * Check if this is a SpiServerPlugin and if so 'collect' it to give the complete list
    * later on the DefaultServer for late call to configure().
@@ -400,7 +429,8 @@ public class InternalConfiguration {
 
     TransactionManagerOptions options =
       new TransactionManagerOptions(notifyL2CacheInForeground, serverConfig, scopeManager, clusterManager, backgroundExecutor,
-                                    indexUpdateProcessor, beanDescriptorManager, dataSource(), profileHandler(), logManager);
+                                    indexUpdateProcessor, beanDescriptorManager, dataSource(), profileHandler(), logManager,
+                                    tableModState, cacheNotify, clockService);
 
     if (serverConfig.isExplicitTransactionBeginMode()) {
       return new ExplicitTransactionManager(options);
@@ -493,7 +523,7 @@ public class InternalConfiguration {
     return dataTimeZone;
   }
 
-  public ServerCacheManager cache() {
+  public ServerCacheManager cacheManager() {
     return new DefaultCacheAdapter(cacheManager);
   }
 
@@ -537,4 +567,62 @@ public class InternalConfiguration {
   public SpiLogManager getLogManager() {
     return logManager;
   }
+
+  private ServerCachePlugin initServerCachePlugin() {
+
+    ServerCachePlugin plugin = serverConfig.getServerCachePlugin();
+    if (plugin == null) {
+      ServiceLoader<ServerCachePlugin> cacheFactories = ServiceLoader.load(ServerCachePlugin.class);
+      Iterator<ServerCachePlugin> iterator = cacheFactories.iterator();
+      if (iterator.hasNext()) {
+        // use the cacheFactory (via classpath service loader)
+        plugin = iterator.next();
+        logger.debug("using ServerCacheFactory {}", plugin.getClass());
+      } else {
+        // use the built in default l2 caching which is local cache based
+        localL2Caching = true;
+        plugin = new DefaultServerCachePlugin();
+      }
+    }
+    return plugin;
+  }
+
+  /**
+   * Create and return the CacheManager.
+   */
+  private SpiCacheManager initCacheManager() {
+
+    if (!online || serverConfig.isDisableL2Cache()) {
+      // use local only L2 cache implementation as placeholder
+      return new DefaultServerCacheManager();
+    }
+
+    ServerCacheFactory factory = serverCachePlugin.create(serverConfig, backgroundExecutor);
+
+    ServerCacheNotifyPlugin notifyPlugin = serverConfig.service(ServerCacheNotifyPlugin.class);
+    if (notifyPlugin != null) {
+      // plugin supplied so use that to send notifications
+      cacheNotify = notifyPlugin.create(serverConfig);
+    } else {
+      cacheNotify = factory.createCacheNotify(tableModState);
+    }
+
+    // reasonable default settings are for a cache per bean type
+    ServerCacheOptions beanOptions = new ServerCacheOptions();
+    beanOptions.setMaxSize(serverConfig.getCacheMaxSize());
+    beanOptions.setMaxIdleSecs(serverConfig.getCacheMaxIdleTime());
+    beanOptions.setMaxSecsToLive(serverConfig.getCacheMaxTimeToLive());
+
+    // reasonable default settings for the query cache per bean type
+    ServerCacheOptions queryOptions = new ServerCacheOptions();
+    queryOptions.setMaxSize(serverConfig.getQueryCacheMaxSize());
+    queryOptions.setMaxIdleSecs(serverConfig.getQueryCacheMaxIdleTime());
+    queryOptions.setMaxSecsToLive(serverConfig.getQueryCacheMaxTimeToLive());
+
+    CacheManagerOptions builder = new CacheManagerOptions(clusterManager, serverConfig, localL2Caching)
+      .with(beanOptions, queryOptions)
+      .with(factory, tableModState);
+
+    return new DefaultServerCacheManager(builder);
+  }
 }

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

@@ -8,6 +8,7 @@ import io.ebean.Version;
 import io.ebean.bean.BeanCollection;
 import io.ebean.bean.EntityBean;
 import io.ebean.bean.PersistenceContext;
+import io.ebean.cache.QueryCacheEntry;
 import io.ebean.common.BeanList;
 import io.ebean.common.CopyOnFirstWriteList;
 import io.ebean.event.BeanFindController;
@@ -83,6 +84,8 @@ public final class OrmQueryRequest<T> extends BeanRequest implements SpiOrmQuery
 
   private boolean inlineCountDistinct;
 
+  private Set<String> dependentTables;
+
   /**
    * Create the InternalQueryRequest.
    */
@@ -207,7 +210,7 @@ public final class OrmQueryRequest<T> extends BeanRequest implements SpiOrmQuery
   /**
    * Prepare the query and calculate the query plan key.
    */
-  public void prepareQuery() {
+  void prepareQuery() {
     beanDescriptor.prepareQuery(query);
     adapterPreQuery();
     this.secondaryQueries = query.convertJoins();
@@ -680,8 +683,10 @@ public final class OrmQueryRequest<T> extends BeanRequest implements SpiOrmQuery
     }
   }
 
-  public void putToQueryCache(Object queryResult) {
-    beanDescriptor.queryCachePut(cacheKey, queryResult);
+  public void putToQueryCache(Object result) {
+    // use transaction start where as query statement start would be better at READ_COMMITTED
+    long asOfTimestamp = transaction.getStartMillis();
+    beanDescriptor.queryCachePut(cacheKey, new QueryCacheEntry(result, dependentTables, asOfTimestamp));
   }
 
   /**
@@ -752,4 +757,13 @@ public final class OrmQueryRequest<T> extends BeanRequest implements SpiOrmQuery
   public boolean isInlineCountDistinct() {
     return inlineCountDistinct;
   }
+
+  public void addDependentTables(Set<String> tables) {
+    if (tables != null && !tables.isEmpty()) {
+      if (dependentTables == null) {
+        dependentTables = new LinkedHashSet<>();
+      }
+      dependentTables.addAll(tables);
+    }
+  }
 }

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

@@ -11,6 +11,7 @@ import io.ebean.bean.BeanCollection;
 import io.ebean.bean.EntityBean;
 import io.ebean.bean.EntityBeanIntercept;
 import io.ebean.bean.PersistenceContext;
+import io.ebean.cache.QueryCacheEntry;
 import io.ebean.config.EncryptKey;
 import io.ebean.config.ServerConfig;
 import io.ebean.config.dbplatform.IdType;
@@ -1361,8 +1362,8 @@ public class BeanDescriptor<T> implements BeanType<T>, STreeType {
   /**
    * Put a query result into the query cache.
    */
-  public void queryCachePut(Object id, Object queryResult) {
-    cacheHelp.queryCachePut(id, queryResult);
+  public void queryCachePut(Object id, QueryCacheEntry entry) {
+    cacheHelp.queryCachePut(id, entry);
   }
 
   /**

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

@@ -4,6 +4,7 @@ import io.ebean.bean.BeanCollection;
 import io.ebean.bean.EntityBean;
 import io.ebean.bean.EntityBeanIntercept;
 import io.ebean.bean.PersistenceContext;
+import io.ebean.cache.QueryCacheEntry;
 import io.ebean.cache.ServerCache;
 import io.ebeaninternal.api.BeanCacheResult;
 import io.ebeaninternal.api.TransactionEventTable.TableIUD;
@@ -54,6 +55,7 @@ final class BeanDescriptorCacheHelp<T> {
    * Flag indicating this bean has no relationships.
    */
   private final boolean cacheSharableBeans;
+  private final boolean invalidateQueryCache;
 
   private final Class<?> beanType;
 
@@ -84,6 +86,7 @@ final class BeanDescriptorCacheHelp<T> {
     this.cacheName = beanType.getSimpleName();
     this.cacheManager = cacheManager;
     this.cacheOptions = cacheOptions;
+    this.invalidateQueryCache = cacheOptions.isInvalidateQueryCache();
     this.cacheSharableBeans = cacheSharableBeans;
     this.propertiesOneImported = propertiesOneImported;
     this.naturalKey = cacheOptions.getNaturalKey();
@@ -111,7 +114,7 @@ final class BeanDescriptorCacheHelp<T> {
    * Derive the cache notify flags.
    */
   void deriveNotifyFlags() {
-    cacheNotifyOnAll = (beanCache != null || queryCache != null);
+    cacheNotifyOnAll = (invalidateQueryCache || beanCache != null || queryCache != null);
     cacheNotifyOnDelete = !cacheNotifyOnAll && isNotifyOnDeletes();
 
     if (logger.isDebugEnabled()) {
@@ -225,14 +228,14 @@ final class BeanDescriptorCacheHelp<T> {
   /**
    * Put a query result into the query cache.
    */
-  void queryCachePut(Object id, Object queryResult) {
+  void queryCachePut(Object id, QueryCacheEntry entry) {
     if (queryCache == null) {
       throw new IllegalStateException("No query cache enabled on " + desc + ". Need explicit @Cache(enableQueryCache=true)");
     }
     if (queryLog.isDebugEnabled()) {
       queryLog.debug("   PUT {}({})", cacheName, id);
     }
-    queryCache.put(id, queryResult);
+    queryCache.put(id, entry);
   }
 
 
@@ -723,30 +726,42 @@ final class BeanDescriptorCacheHelp<T> {
    * Add appropriate cache changes to support delete by id.
    */
   void handleDelete(Object id, CacheChangeSet changeSet) {
-    if (beanCache != null) {
-      changeSet.addBeanRemove(desc, id);
+    if (invalidateQueryCache) {
+     changeSet.addInvalidate(desc);
+    } else {
+      if (beanCache != null) {
+        changeSet.addBeanRemove(desc, id);
+      }
+      cacheDeleteImported(true, null, changeSet);
     }
-    cacheDeleteImported(true, null, changeSet);
   }
 
   /**
    * Add appropriate cache changes to support delete bean.
    */
   void handleDelete(Object id, PersistRequestBean<T> deleteRequest, CacheChangeSet changeSet) {
-    queryCacheClear(changeSet);
-    if (beanCache != null) {
-      changeSet.addBeanRemove(desc, id);
+    if (invalidateQueryCache) {
+      changeSet.addInvalidate(desc);
+    } else {
+      queryCacheClear(changeSet);
+      if (beanCache != null) {
+        changeSet.addBeanRemove(desc, id);
+      }
+      cacheDeleteImported(true, deleteRequest.getEntityBean(), changeSet);
     }
-    cacheDeleteImported(true, deleteRequest.getEntityBean(), changeSet);
   }
 
   /**
    * Add appropriate cache changes to support insert.
    */
   void handleInsert(PersistRequestBean<T> insertRequest, CacheChangeSet changeSet) {
-    queryCacheClear(changeSet);
-    cacheDeleteImported(false, insertRequest.getEntityBean(), changeSet);
-    changeSet.addBeanInsert(desc.getBaseTable());
+    if (invalidateQueryCache) {
+      changeSet.addInvalidate(desc);
+    } else {
+      queryCacheClear(changeSet);
+      cacheDeleteImported(false, insertRequest.getEntityBean(), changeSet);
+      changeSet.addBeanInsert(desc.getBaseTable());
+    }
   }
 
   private void cacheDeleteImported(boolean clear, EntityBean entityBean, CacheChangeSet changeSet) {
@@ -759,28 +774,30 @@ final class BeanDescriptorCacheHelp<T> {
    * Add appropriate changes to support update.
    */
   void handleUpdate(Object id, PersistRequestBean<T> updateRequest, CacheChangeSet changeSet) {
+    if (invalidateQueryCache) {
+      changeSet.addInvalidate(desc);
 
-    queryCacheClear(changeSet);
+    } else {
+      queryCacheClear(changeSet);
+      if (beanCache == null) {
+        // query caching only
+        return;
+      }
 
-    if (beanCache == null) {
-      // query caching only
-      return;
-    }
-
-    List<BeanPropertyAssocMany<?>> manyCollections = updateRequest.getUpdatedManyCollections();
-    if (manyCollections != null) {
-      for (BeanPropertyAssocMany<?> many : manyCollections) {
-        if (!many.isElementCollection()) {
-          Object details = many.getValue(updateRequest.getEntityBean());
-          CachedManyIds entry = createManyIds(many, details);
-          if (entry != null) {
-            changeSet.addManyPut(desc, many.getName(), id, entry);
+      List<BeanPropertyAssocMany<?>> manyCollections = updateRequest.getUpdatedManyCollections();
+      if (manyCollections != null) {
+        for (BeanPropertyAssocMany<?> many : manyCollections) {
+          if (!many.isElementCollection()) {
+            Object details = many.getValue(updateRequest.getEntityBean());
+            CachedManyIds entry = createManyIds(many, details);
+            if (entry != null) {
+              changeSet.addManyPut(desc, many.getName(), id, entry);
+            }
           }
         }
       }
+      updateRequest.addBeanUpdate(changeSet);
     }
-
-    updateRequest.addBeanUpdate(changeSet);
   }
 
   /**

src/main/java/io/ebeaninternal/server/deploy/meta/DeployBeanDescriptor.java

@@ -440,6 +440,13 @@ public class DeployBeanDescriptor<T> {
     this.inheritInfo = inheritInfo;
   }
 
+  /**
+   * Set that this type invalidates query caches.
+   */
+  public void setInvalidateQueryCache() {
+    this.cacheOptions = CacheOptions.INVALIDATE_QUERY_CACHE;
+  }
+
   /**
    * Enable L2 bean and query caching based on Cache annotation.
    */

src/main/java/io/ebeaninternal/server/deploy/parse/AnnotationClass.java

@@ -7,6 +7,7 @@ import io.ebean.annotation.Draftable;
 import io.ebean.annotation.DraftableElement;
 import io.ebean.annotation.History;
 import io.ebean.annotation.Index;
+import io.ebean.annotation.InvalidateQueryCache;
 import io.ebean.annotation.ReadAudit;
 import io.ebean.annotation.UpdateMode;
 import io.ebean.annotation.View;
@@ -181,9 +182,16 @@ public class AnnotationClass extends AnnotationParser {
       descriptor.setUpdateChangesOnly(updateMode.updateChangesOnly());
     }
 
-    Cache cache = AnnotationUtil.findAnnotationRecursive(cls, Cache.class);
-    if (cache != null && !disableL2Cache) {
-      descriptor.setCache(cache);
+    if (!disableL2Cache) {
+      Cache cache = AnnotationUtil.findAnnotationRecursive(cls, Cache.class);
+      if (cache != null) {
+        descriptor.setCache(cache);
+      } else {
+        InvalidateQueryCache invalidateQueryCache = AnnotationUtil.findAnnotationRecursive(cls, InvalidateQueryCache.class);
+        if (invalidateQueryCache != null) {
+          descriptor.setInvalidateQueryCache();
+        }
+      }
     }
 
     Set<NamedQuery> namedQueries = AnnotationUtil.findAnnotationsRecursive(cls, NamedQuery.class);

src/main/java/io/ebeaninternal/server/query/CQuery.java

@@ -38,6 +38,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
+import java.util.Set;
 
 /**
  * An object that represents a SqlSelect statement.
@@ -811,4 +812,8 @@ public class CQuery<T> implements DbReadContext, CancelableQuery, SpiProfileTran
   PreparedStatement getPstmt() {
     return pstmt;
   }
+
+  public Set<String> getDependentTables() {
+    return queryPlan.getDependentTables();
+  }
 }

src/main/java/io/ebeaninternal/server/query/CQueryEngine.java

@@ -25,6 +25,8 @@ import org.slf4j.LoggerFactory;
 import javax.persistence.PersistenceException;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -111,6 +113,15 @@ public class CQueryEngine {
       if (request.isLogSummary()) {
         request.getTransaction().logSummary(rcQuery.getSummary());
       }
+      if (request.isQueryCachePut() && !list.isEmpty()) {
+        request.addDependentTables(rcQuery.getDependentTables());
+
+        list = Collections.unmodifiableList(list);
+        request.putToQueryCache(list);
+        if (Boolean.FALSE.equals(request.getQuery().isReadOnly())) {
+          list = new ArrayList<>(list);
+        }
+      }
       return list;
 
     } catch (SQLException e) {
@@ -173,6 +184,11 @@ public class CQueryEngine {
         request.getTransaction().end();
       }
 
+      if (request.isQueryCachePut()) {
+        request.addDependentTables(rcQuery.getDependentTables());
+        request.putToQueryCache(count);
+      }
+
       return count;
 
     } catch (SQLException e) {
@@ -385,6 +401,9 @@ public class CQueryEngine {
       }
 
       request.executeSecondaryQueries(false);
+      if (request.isQueryCachePut()) {
+        request.addDependentTables(cquery.getDependentTables());
+      }
 
       return beanCollection;
 

src/main/java/io/ebeaninternal/server/query/CQueryFetchSingleAttribute.java

@@ -17,6 +17,7 @@ import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
 
 /**
  * Base compiled query request for single attribute queries.
@@ -188,4 +189,8 @@ class CQueryFetchSingleAttribute implements SpiProfileTransactionEvent {
       .profileStream()
       .addQueryEvent(query.profileEventId(), profileOffset, desc.getProfileId(), rowCount, query.getProfileId());
   }
+
+  Set<String> getDependentTables() {
+    return queryPlan.getDependentTables();
+  }
 }

src/main/java/io/ebeaninternal/server/query/CQueryPlan.java

@@ -24,6 +24,8 @@ import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.util.Collections;
+import java.util.Set;
 
 /**
  * Represents a query for a given SQL statement.
@@ -88,6 +90,8 @@ public class CQueryPlan {
    */
   private volatile String auditQueryHash;
 
+  private final Set<String> dependentTables;
+
   /**
    * Create a query plan based on a OrmQueryRequest.
    */
@@ -110,6 +114,7 @@ public class CQueryPlan {
     this.logWhereSql = logWhereSql;
     this.encryptedProps = sqlTree.getEncryptedProps();
     this.stats = new CQueryPlanStats(this, server.isCollectQueryOrigins());
+    this.dependentTables = sqlTree.dependentTables();
   }
 
   /**
@@ -134,6 +139,7 @@ public class CQueryPlan {
     this.logWhereSql = logWhereSql;
     this.encryptedProps = sqlTree.getEncryptedProps();
     this.stats = new CQueryPlanStats(this, server.isCollectQueryOrigins());
+    this.dependentTables = (rawSql) ? Collections.emptySet() : sqlTree.dependentTables();
   }
 
   private String location() {
@@ -154,6 +160,10 @@ public class CQueryPlan {
     return beanType;
   }
 
+  public Set<String> getDependentTables() {
+    return dependentTables;
+  }
+
   public ProfileLocation getProfileLocation() {
     return profileLocation;
   }

src/main/java/io/ebeaninternal/server/query/CQueryRowCount.java

@@ -12,6 +12,7 @@ import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.util.Set;
 
 /**
  * Executes the select row count query.
@@ -154,4 +155,8 @@ class CQueryRowCount implements SpiProfileTransactionEvent {
       .profileStream()
       .addQueryEvent(query.profileEventId(), profileOffset, desc.getProfileId(), rowCount, query.getProfileId());
   }
+
+  Set<String> getDependentTables() {
+    return queryPlan.getDependentTables();
+  }
 }

src/main/java/io/ebeaninternal/server/query/DefaultOrmQueryEngine.java

@@ -88,41 +88,20 @@ public class DefaultOrmQueryEngine implements OrmQueryEngine {
   public <T> int findCount(OrmQueryRequest<T> request) {
 
     flushJdbcBatchOnQuery(request);
-    int result = queryEngine.findCount(request);
-    if (request.getQuery().getUseQueryCache().isPut()) {
-      request.putToQueryCache(result);
-    }
-    return result;
+    return queryEngine.findCount(request);
   }
 
   @Override
   public <A> List<A> findIds(OrmQueryRequest<?> request) {
 
     flushJdbcBatchOnQuery(request);
-    List<A> result = queryEngine.findIds(request);
-    if (request.getQuery().getUseQueryCache().isPut()) {
-      result = Collections.unmodifiableList(result);
-      request.putToQueryCache(result);
-      if (Boolean.FALSE.equals(request.getQuery().isReadOnly())) {
-        result = new ArrayList<>(result);
-      }
-    }
-    return result;
+    return queryEngine.findIds(request);
   }
 
   @Override
   public <A> List<A> findSingleAttributeList(OrmQueryRequest<?> request) {
     flushJdbcBatchOnQuery(request);
-    List<A> result = queryEngine.findSingleAttributeList(request);
-    if (!result.isEmpty() && request.getQuery().getUseQueryCache().isPut()) {
-      // load the query result into the query cache
-      result = Collections.unmodifiableList(result);
-      request.putToQueryCache(result);
-      if (Boolean.FALSE.equals(request.getQuery().isReadOnly())) {
-        result = new ArrayList<>(result);
-      }
-    }
-    return result;
+    return queryEngine.findSingleAttributeList(request);
   }
 
   @Override

src/main/java/io/ebeaninternal/server/query/SqlTree.java

@@ -3,6 +3,7 @@ package io.ebeaninternal.server.query;
 import io.ebeaninternal.api.SpiQuery;
 
 import java.util.ArrayList;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
 
@@ -163,4 +164,13 @@ class SqlTree {
   boolean isSingleProperty() {
     return rootNode.isSingleProperty();
   }
+
+  /**
+   * Return the tables that are joined in this query.
+   */
+  Set<String> dependentTables() {
+    Set<String> tables = new LinkedHashSet<>();
+    rootNode.dependentTables(tables);
+    return tables;
+  }
 }

src/main/java/io/ebeaninternal/server/query/SqlTreeNode.java

@@ -9,6 +9,7 @@ import io.ebeaninternal.server.type.ScalarType;
 
 import java.sql.SQLException;
 import java.util.List;
+import java.util.Set;
 
 interface SqlTreeNode {
 
@@ -87,4 +88,9 @@ interface SqlTreeNode {
    * Return true if the query is known to only have a single property selected.
    */
   boolean isSingleProperty();
+
+  /**
+   * Add dependent tables to the given set.
+   */
+  void dependentTables(Set<String> tables);
 }

src/main/java/io/ebeaninternal/server/query/SqlTreeNodeBean.java

@@ -21,6 +21,7 @@ import java.sql.Timestamp;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 /**
  * Normal bean included in the query.
@@ -578,6 +579,14 @@ class SqlTreeNodeBean implements SqlTreeNode {
     }
   }
 
+  @Override
+  public void dependentTables(Set<String> tables) {
+    tables.add(nodeBeanProp.target().getBaseTable(temporalMode));
+    for (SqlTreeNode child : children) {
+      child.dependentTables(tables);
+    }
+  }
+
   /**
    * Join to base table for this node. This includes a join to the intersection
    * table if this is a ManyToMany node.

src/main/java/io/ebeaninternal/server/query/SqlTreeNodeExtraJoin.java

@@ -11,6 +11,7 @@ import io.ebeaninternal.server.type.ScalarType;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
 
 /**
  * The purpose is to add an extra join to the query.
@@ -100,6 +101,16 @@ class SqlTreeNodeExtraJoin implements SqlTreeNode {
     children.add(child);
   }
 
+  @Override
+  public void dependentTables(Set<String> tables) {
+    tables.add(assocBeanProperty.target().getBaseTable(SpiQuery.TemporalMode.CURRENT));
+    if (children != null) {
+      for (SqlTreeNode child : children) {
+        child.dependentTables(tables);
+      }
+    }
+  }
+
   @Override
   public void appendFrom(DbSqlContext ctx, SqlJoinType joinType) {
 

src/main/java/io/ebeaninternal/server/query/SqlTreeNodeManyRoot.java

@@ -7,6 +7,7 @@ import io.ebeaninternal.server.deploy.DbSqlContext;
 
 import java.sql.SQLException;
 import java.util.List;
+import java.util.Set;
 
 final class SqlTreeNodeManyRoot extends SqlTreeNodeBean {
 

src/main/java/io/ebeaninternal/server/query/SqlTreeNodeManyWhereJoin.java

@@ -10,6 +10,7 @@ import io.ebeaninternal.server.deploy.TableJoin;
 import io.ebeaninternal.server.type.ScalarType;
 
 import java.util.List;
+import java.util.Set;
 
 /**
  * Join to Many (or child of a many) to support where clause predicates on many properties.
@@ -109,6 +110,11 @@ class SqlTreeNodeManyWhereJoin implements SqlTreeNode {
     }
   }
 
+  @Override
+  public void dependentTables(Set<String> tables) {
+    tables.add(nodeBeanProp.target().getBaseTable(SpiQuery.TemporalMode.CURRENT));
+  }
+
   @Override
   public void buildRawSqlSelectChain(List<String> selectChain) {
     // nothing to add

src/main/java/io/ebeaninternal/server/query/SqlTreeNodeRoot.java

@@ -5,6 +5,7 @@ import io.ebeaninternal.server.deploy.DbSqlContext;
 import io.ebeaninternal.server.deploy.TableJoin;
 
 import java.util.List;
+import java.util.Set;
 
 /**
  * Represents the root node of the Sql Tree.
@@ -77,4 +78,10 @@ final class SqlTreeNodeRoot extends SqlTreeNodeBean {
     return joinType;
   }
 
+  @Override
+  public void dependentTables(Set<String> tables) {
+    for (SqlTreeNode child : children) {
+      child.dependentTables(tables);
+    }
+  }
 }

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

@@ -1,8 +1,8 @@
 package io.ebeaninternal.server.transaction;
 
 import io.ebeaninternal.api.SpiEbeanServer;
-import io.ebeaninternal.server.cluster.BinaryMessage;
-import io.ebeaninternal.server.cluster.BinaryMessageList;
+import io.ebeaninternal.server.cluster.binarymessage.BinaryMessage;
+import io.ebeaninternal.server.cluster.binarymessage.BinaryMessageList;
 import io.ebeaninternal.server.core.PersistRequest;
 import io.ebeaninternal.server.deploy.BeanDescriptor;
 import io.ebeaninternal.server.deploy.id.IdBinder;
@@ -140,7 +140,7 @@ public class BeanPersistIds {
           idBinder.writeData(os, idList.get(i));
         }
 
-        os.flush();
+        os.close();
         msgList.add(m);
 
       } while (i < eof);
@@ -167,7 +167,7 @@ public class BeanPersistIds {
     return sb.toString();
   }
 
-  void addId(PersistRequest.Type type, Serializable id) {
+  public void addId(PersistRequest.Type type, Serializable id) {
     switch (type) {
       case INSERT:
         addInsertId(id);
@@ -210,10 +210,18 @@ public class BeanPersistIds {
     return beanDescriptor;
   }
 
-  List<Object> getDeleteIds() {
+  public List<Object> getDeleteIds() {
     return deleteIds;
   }
 
+  public List<Object> getInsertIds() {
+    return insertIds;
+  }
+
+  public List<Object> getUpdateIds() {
+    return updateIds;
+  }
+
   /**
    * Notify the cache of this event that came from another server in the cluster.
    */

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

@@ -67,6 +67,7 @@ class ImplicitReadOnlyTransaction implements SpiTransaction, TxnProfileEventCode
   private Map<String, Object> userObjects;
 
   private long startNanos;
+  private long startMillis;
 
   /**
    * Create without a tenantId.
@@ -79,6 +80,7 @@ class ImplicitReadOnlyTransaction implements SpiTransaction, TxnProfileEventCode
     this.connection = connection;
     this.persistenceContext = new DefaultPersistenceContext();
     this.startNanos = System.nanoTime();
+    this.startMillis = manager.clockNowMillis();
   }
 
   /**
@@ -89,6 +91,12 @@ class ImplicitReadOnlyTransaction implements SpiTransaction, TxnProfileEventCode
     this.tenantId = tenantId;
   }
 
+  @Override
+  public long getStartMillis() {
+    // not used on read only transaction
+    return startMillis;
+  }
+
   @Override
   public void setLabel(String label) {
     // do nothing

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

@@ -187,6 +187,7 @@ public class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes {
   protected ProfileLocation profileLocation;
 
   protected final long startNanos;
+  private final long startMillis;
 
   /**
    * Create a new JdbcTransaction.
@@ -203,6 +204,7 @@ public class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes {
       this.startNanos = System.nanoTime();
 
       if (manager == null) {
+        this.startMillis = System.currentTimeMillis();
         this.logSql = false;
         this.logSummary = false;
         this.skipCacheAfterWrite = true;
@@ -210,6 +212,7 @@ public class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes {
         this.batchOnCascadeMode = PersistBatch.NONE;
         this.onQueryOnly = OnQueryOnly.ROLLBACK;
       } else {
+        this.startMillis = manager.clockNowMillis();
         this.logSql = manager.isLogSql();
         this.logSummary = manager.isLogSummary();
         this.skipCacheAfterWrite = manager.isSkipCacheAfterWrite();
@@ -235,6 +238,11 @@ public class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes {
     return label;
   }
 
+  @Override
+  public long getStartMillis() {
+    return startMillis;
+  }
+
   @Override
   public long profileOffset() {
     return (profileStream == null) ? 0 : profileStream.offset();
@@ -853,7 +861,7 @@ public class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes {
   public TransactionEvent getEvent() {
     queryOnly = false;
     if (event == null) {
-      event = new TransactionEvent();
+      event = new TransactionEvent(startMillis);
     }
     return event;
   }
@@ -1050,7 +1058,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/NoTransaction.java

@@ -47,6 +47,12 @@ class NoTransaction implements SpiTransaction {
 
   }
 
+  @Override
+  public long getStartMillis() {
+    // not used
+    return System.currentTimeMillis();
+  }
+
   @Override
   public boolean isActive() {
     // always false

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

@@ -14,6 +14,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
+import java.util.Set;
 
 /**
  * Performs post commit processing using a background thread.
@@ -21,7 +22,7 @@ import java.util.List;
  * This includes Cluster notification, and BeanPersistListeners.
  * </p>
  */
-public final class PostCommitProcessing {
+final class PostCommitProcessing {
 
   private static final Logger logger = LoggerFactory.getLogger(PostCommitProcessing.class);
 
@@ -50,7 +51,7 @@ public final class PostCommitProcessing {
   /**
    * Create for an external modification.
    */
-  public PostCommitProcessing(ClusterManager clusterManager, TransactionManager manager, TransactionEvent event) {
+  PostCommitProcessing(ClusterManager clusterManager, TransactionManager manager, TransactionEvent event) {
 
     this.clusterManager = clusterManager;
     this.manager = manager;
@@ -67,7 +68,7 @@ public final class PostCommitProcessing {
   /**
    * Create for a transaction.
    */
-  public PostCommitProcessing(ClusterManager clusterManager, TransactionManager manager, SpiTransaction transaction) {
+  PostCommitProcessing(ClusterManager clusterManager, TransactionManager manager, SpiTransaction transaction) {
 
     this.clusterManager = clusterManager;
     this.manager = manager;
@@ -88,10 +89,10 @@ public final class PostCommitProcessing {
     processTableEvents(event.getEventTables());
     if (manager.notifyL2CacheInForeground) {
       // process l2 cache changes in foreground
-      processCacheChanges(event.buildCacheChanges(manager.viewInvalidation));
+      processCacheChanges(event.buildCacheChanges());
     } else {
       // collect l2 cache changes for delayed background processing
-      cacheChanges = event.buildCacheChanges(manager.viewInvalidation);
+      cacheChanges = event.buildCacheChanges();
     }
   }
 
@@ -165,7 +166,12 @@ public final class PostCommitProcessing {
    */
   private void processCacheChanges(CacheChangeSet cacheChanges) {
     if (cacheChanges != null) {
-      manager.processViewInvalidation(cacheChanges.apply());
+      Set<String> touched = cacheChanges.touchedTables();
+      if (touched != null && !touched.isEmpty()) {
+        manager.processTouchedTables(touched, cacheChanges.modificationTimestamp());
+        // TODO: Propagate touched tables to other cluster members
+      }
+      cacheChanges.apply();
     }
   }
 
@@ -205,6 +211,10 @@ public final class PostCommitProcessing {
 
     RemoteTransactionEvent remoteTransactionEvent = new RemoteTransactionEvent(serverName);
 
+    Set<String> touched = cacheChanges.touchedTables();
+    if (touched != null && !touched.isEmpty()) {
+      remoteTransactionEvent.addRemoteTableMod(new RemoteTableMod(cacheChanges.modificationTimestamp(), touched));
+    }
     if (beanPersistIdMap != null) {
       for (BeanPersistIds beanPersist : beanPersistIdMap.values()) {
         remoteTransactionEvent.addBeanPersistIds(beanPersist);

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

@@ -0,0 +1,55 @@
+package io.ebeaninternal.server.transaction;
+
+import io.ebeaninternal.server.cluster.binarymessage.BinaryMessage;
+import io.ebeaninternal.server.cluster.binarymessage.BinaryMessageList;
+
+import java.io.DataInput;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+public class RemoteTableMod {
+
+  private final long timestamp;
+
+  private final Set<String> tables;
+
+  public RemoteTableMod(long timestamp, Set<String> tables) {
+    this.timestamp = timestamp;
+    this.tables = tables;
+  }
+
+  public long getTimestamp() {
+    return timestamp;
+  }
+
+  public Set<String> getTables() {
+    return tables;
+  }
+
+  public static RemoteTableMod readBinaryMessage(DataInput dataInput) throws IOException {
+
+    long timestamp = dataInput.readLong();
+    int count = dataInput.readInt();
+
+    Set<String> tables = new LinkedHashSet<>();
+    for (int i = 0; i < count; i++) {
+      tables.add(dataInput.readUTF());
+    }
+    return new RemoteTableMod(timestamp, tables);
+  }
+
+  public void writeBinary(BinaryMessageList msgList) throws IOException {
+    BinaryMessage msg = new BinaryMessage(tables.size() * 20  + 30);
+    DataOutputStream os = msg.getOs();
+    os.writeInt(BinaryMessage.TYPE_TABLEMOD);
+    os.writeLong(timestamp);
+    os.writeInt(tables.size());
+    for (String table : tables) {
+      os.writeUTF(table);
+    }
+    os.close();
+    msgList.add(msg);
+  }
+}

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

@@ -3,7 +3,7 @@ package io.ebeaninternal.server.transaction;
 import io.ebeaninternal.api.SpiEbeanServer;
 import io.ebeaninternal.api.TransactionEventTable.TableIUD;
 import io.ebeaninternal.server.cache.RemoteCacheEvent;
-import io.ebeaninternal.server.cluster.BinaryMessageList;
+import io.ebeaninternal.server.cluster.binarymessage.BinaryMessageList;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -19,6 +19,8 @@ public class RemoteTransactionEvent implements Runnable {
 
   private RemoteCacheEvent remoteCacheEvent;
 
+  private RemoteTableMod remoteTableMod;
+
   private String serverName;
 
   private transient SpiEbeanServer server;
@@ -53,6 +55,10 @@ public class RemoteTransactionEvent implements Runnable {
 
   public void writeBinaryMessage(BinaryMessageList msgList) throws IOException {
 
+    if (remoteTableMod != null) {
+      remoteTableMod.writeBinary(msgList);
+    }
+
     if (tableList != null) {
       for (TableIUD aTableList : tableList) {
         aTableList.writeBinaryMessage(msgList);
@@ -113,6 +119,10 @@ public class RemoteTransactionEvent implements Runnable {
     tableList.add(tableIud);
   }
 
+  public void addRemoteTableMod(RemoteTableMod remoteTableMod) {
+    this.remoteTableMod = remoteTableMod;
+  }
+
   public String getServerName() {
     return serverName;
   }
@@ -141,4 +151,8 @@ public class RemoteTransactionEvent implements Runnable {
     return remoteCacheEvent;
   }
 
+  public RemoteTableMod getRemoteTableMod() {
+    return remoteTableMod;
+  }
+
 }

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

@@ -0,0 +1,70 @@
+package io.ebeaninternal.server.transaction;
+
+import io.ebean.cache.QueryCacheEntry;
+import io.ebean.cache.QueryCacheEntryValidate;
+import io.ebean.cache.ServerCacheNotification;
+import io.ebean.cache.ServerCacheNotify;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Holds timestamp of last modification per table.
+ * <p>
+ * This information is used to validate entries in the L2 query caches.
+ * </p>
+ */
+public class TableModState implements QueryCacheEntryValidate, ServerCacheNotify {
+
+  private static final Logger log = LoggerFactory.getLogger("io.ebean.cache.TABLEMOD");
+
+  private Map<String,Long> tableModStamp = new ConcurrentHashMap<>();
+
+  /**
+   * Set the modified timestamp on the tables that have been touched.
+   */
+  void touch(Set<String> touchedTables, long modTimestamp) {
+    for (String tableName : touchedTables) {
+      tableModStamp.put(tableName, modTimestamp);
+    }
+    if (log.isDebugEnabled()) {
+      log.debug("TableModState updated - " + tableModStamp);
+    }
+  }
+
+  /**
+   * Return true if all the tables are valid based on timestamp comparison.
+   */
+  boolean isValid(Set<String> tables, long sinceTimestamp) {
+    for (String tableName : tables) {
+      Long modTime = tableModStamp.get(tableName);
+      if (modTime != null && modTime > sinceTimestamp ) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  @Override
+  public boolean isValid(QueryCacheEntry entry) {
+    Set<String> dependentTables = entry.getDependentTables();
+    if (dependentTables != null && !dependentTables.isEmpty()) {
+      return isValid(dependentTables, entry.getTimestamp());
+    }
+    return true;
+  }
+
+  /**
+   * Update the table modification timestamps based on remote table modification events.
+   */
+  @Override
+  public void notify(ServerCacheNotification notification) {
+
+    // TODO: Change to use ClockService ...
+    long modifyTimestamp = notification.getModifyTimestamp();
+    touch(notification.getDependentTables(), modifyTimestamp);
+  }
+}

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

@@ -5,6 +5,8 @@ import io.ebean.ProfileLocation;
 import io.ebean.TxScope;
 import io.ebean.annotation.PersistBatch;
 import io.ebean.annotation.TxType;
+import io.ebean.cache.ServerCacheNotification;
+import io.ebean.cache.ServerCacheNotify;
 import io.ebean.config.CurrentTenantProvider;
 import io.ebean.config.dbplatform.DatabasePlatform;
 import io.ebean.config.dbplatform.DatabasePlatform.OnQueryOnly;
@@ -27,6 +29,7 @@ import io.ebeaninternal.metric.MetricFactory;
 import io.ebeaninternal.metric.TimedMetric;
 import io.ebeaninternal.metric.TimedMetricMap;
 import io.ebeaninternal.server.cluster.ClusterManager;
+import io.ebeaninternal.server.core.ClockService;
 import io.ebeaninternal.server.deploy.BeanDescriptorManager;
 import io.ebeaninternal.server.profile.TimedProfileLocation;
 import io.ebeaninternal.server.profile.TimedProfileLocationRegistry;
@@ -46,7 +49,7 @@ import java.util.Set;
 /**
  * Manages transactions.
  * <p>
- * Keeps the Cache and Cluster in synch when transactions are committed.
+ * Keeps the Cache and Cluster in sync when transactions are committed.
  * </p>
  */
 public class TransactionManager implements SpiTransactionManager {
@@ -136,6 +139,10 @@ public class TransactionManager implements SpiTransactionManager {
   private final TimedMetricMap txnNamed;
   private final TransactionScopeManager scopeManager;
 
+  private final TableModState tableModState;
+  private final ServerCacheNotify cacheNotify;
+  private final ClockService clockService;
+
   /**
    * Create the TransactionManager
    */
@@ -157,6 +164,9 @@ public class TransactionManager implements SpiTransactionManager {
     this.clusterManager = options.clusterManager;
     this.serverName = options.config.getName();
     this.scopeManager = options.scopeManager;
+    this.tableModState = options.tableModState;
+    this.cacheNotify = options.cacheNotify;
+    this.clockService = options.clockService;
     this.backgroundExecutor = options.backgroundExecutor;
     this.dataSourceSupplier = options.dataSourceSupplier;
     this.docStoreActive = options.config.getDocStoreConfig().isActive();
@@ -178,6 +188,13 @@ public class TransactionManager implements SpiTransactionManager {
     scopeManager.register(this);
   }
 
+  /**
+   * Return the NOW timestamp in epoch millis.
+   */
+  public long clockNowMillis() {
+    return clockService.nowMillis();
+  }
+
   /**
    * Create a new scoped transaction.
    */
@@ -434,7 +451,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);
@@ -495,10 +512,12 @@ public class TransactionManager implements SpiTransactionManager {
   /**
    * Invalidate the query caches for entities based on views.
    */
-  public void processViewInvalidation(Set<String> viewInvalidation) {
-    if (!viewInvalidation.isEmpty()) {
-      beanDescriptorManager.processViewInvalidation(viewInvalidation);
+  public void processTouchedTables(Set<String> touchedTables, long modTimestamp) {
+    tableModState.touch(touchedTables, modTimestamp);
+    if (viewInvalidation) {
+      beanDescriptorManager.processViewInvalidation(touchedTables);
     }
+    cacheNotify.notify(new ServerCacheNotification(modTimestamp, touchedTables));
   }
 
   /**

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

@@ -1,10 +1,12 @@
 package io.ebeaninternal.server.transaction;
 
 import io.ebean.BackgroundExecutor;
+import io.ebean.cache.ServerCacheNotify;
 import io.ebean.config.ServerConfig;
 import io.ebeaninternal.api.SpiLogManager;
 import io.ebeaninternal.api.SpiProfileHandler;
 import io.ebeaninternal.server.cluster.ClusterManager;
+import io.ebeaninternal.server.core.ClockService;
 import io.ebeaninternal.server.deploy.BeanDescriptorManager;
 import io.ebeanservice.docstore.api.DocStoreUpdateProcessor;
 
@@ -24,10 +26,15 @@ public class TransactionManagerOptions {
   final SpiProfileHandler profileHandler;
   final TransactionScopeManager scopeManager;
   final SpiLogManager logManager;
+  final TableModState tableModState;
+  final ServerCacheNotify cacheNotify;
+  final ClockService clockService;
+
 
   public TransactionManagerOptions(boolean notifyL2CacheInForeground, ServerConfig config, TransactionScopeManager scopeManager, ClusterManager clusterManager,
                                    BackgroundExecutor backgroundExecutor, DocStoreUpdateProcessor docStoreUpdateProcessor,
-                                   BeanDescriptorManager descMgr, DataSourceSupplier dataSourceSupplier, SpiProfileHandler profileHandler, SpiLogManager logManager) {
+                                   BeanDescriptorManager descMgr, DataSourceSupplier dataSourceSupplier, SpiProfileHandler profileHandler,
+                                   SpiLogManager logManager, TableModState tableModState, ServerCacheNotify cacheNotify, ClockService clockService) {
 
     this.notifyL2CacheInForeground = notifyL2CacheInForeground;
     this.config = config;
@@ -39,6 +46,9 @@ public class TransactionManagerOptions {
     this.dataSourceSupplier = dataSourceSupplier;
     this.profileHandler = profileHandler;
     this.logManager = logManager;
+    this.tableModState = tableModState;
+    this.cacheNotify = cacheNotify;
+    this.clockService = clockService;
   }
 
 }