#1427 - QueryCache should be cleared, if one of a dependent bean is updated

Propagation across cluster.

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

@@ -10,4 +10,13 @@ public interface ServerCacheFactory {
    */
   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/server/cache/DefaultServerCacheFactory.java

@@ -4,6 +4,8 @@ import io.ebean.BackgroundExecutor;
 import io.ebean.cache.ServerCache;
 import io.ebean.cache.ServerCacheConfig;
 import io.ebean.cache.ServerCacheFactory;
+import io.ebean.cache.ServerCacheNotification;
+import io.ebean.cache.ServerCacheNotify;
 
 
 /**
@@ -16,14 +18,14 @@ 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;
   }
 
@@ -43,4 +45,16 @@ class DefaultServerCacheFactory implements ServerCacheFactory {
     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/core/InternalConfiguration.java

@@ -5,6 +5,8 @@ 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;
@@ -138,6 +140,8 @@ public class InternalConfiguration {
 
   private final ServerCachePlugin serverCachePlugin;
 
+  private ServerCacheNotify cacheNotify;
+
   private boolean localL2Caching;
 
   private final ExpressionFactory expressionFactory;
@@ -418,7 +422,8 @@ public class InternalConfiguration {
 
     TransactionManagerOptions options =
       new TransactionManagerOptions(notifyL2CacheInForeground, serverConfig, scopeManager, clusterManager, backgroundExecutor,
-                                    indexUpdateProcessor, beanDescriptorManager, dataSource(), profileHandler(), logManager, tableModState);
+                                    indexUpdateProcessor, beanDescriptorManager, dataSource(), profileHandler(), logManager,
+                                    tableModState, cacheNotify);
 
     if (serverConfig.isExplicitTransactionBeginMode()) {
       return new ExplicitTransactionManager(options);
@@ -565,7 +570,7 @@ public class InternalConfiguration {
       if (iterator.hasNext()) {
         // use the cacheFactory (via classpath service loader)
         plugin = iterator.next();
-        logger.debug("using ServerCacheFactory {}", serverCachePlugin.getClass());
+        logger.debug("using ServerCacheFactory {}", plugin.getClass());
       } else {
         // use the built in default l2 caching which is local cache based
         localL2Caching = true;
@@ -587,6 +592,14 @@ public class InternalConfiguration {
 
     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());

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

@@ -22,7 +22,7 @@ import java.util.Set;
  * This includes Cluster notification, and BeanPersistListeners.
  * </p>
  */
-public final class PostCommitProcessing {
+final class PostCommitProcessing {
 
   private static final Logger logger = LoggerFactory.getLogger(PostCommitProcessing.class);
 
@@ -51,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;
@@ -68,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;

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

@@ -2,6 +2,10 @@ 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;
@@ -13,24 +17,28 @@ import java.util.concurrent.ConcurrentHashMap;
  * This information is used to validate entries in the L2 query caches.
  * </p>
  */
-public class TableModState implements QueryCacheEntryValidate {
+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.
    */
-  public void touch(Set<String> touchedTables, long modTimestamp) {
-
+  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.
    */
-  public boolean isValid(Set<String> tables, long sinceTimestamp) {
+  boolean isValid(Set<String> tables, long sinceTimestamp) {
     for (String tableName : tables) {
       Long modTime = tableModStamp.get(tableName);
       if (modTime != null && modTime > sinceTimestamp ) {
@@ -48,4 +56,15 @@ public class TableModState implements QueryCacheEntryValidate {
     }
     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;
@@ -137,6 +139,7 @@ public class TransactionManager implements SpiTransactionManager {
   private final TransactionScopeManager scopeManager;
 
   private final TableModState tableModState;
+  private final ServerCacheNotify cacheNotify;
 
   /**
    * Create the TransactionManager
@@ -160,6 +163,7 @@ public class TransactionManager implements SpiTransactionManager {
     this.serverName = options.config.getName();
     this.scopeManager = options.scopeManager;
     this.tableModState = options.tableModState;
+    this.cacheNotify = options.cacheNotify;
     this.backgroundExecutor = options.backgroundExecutor;
     this.dataSourceSupplier = options.dataSourceSupplier;
     this.docStoreActive = options.config.getDocStoreConfig().isActive();
@@ -510,6 +514,7 @@ public class TransactionManager implements SpiTransactionManager {
     if (viewInvalidation) {
       beanDescriptorManager.processViewInvalidation(touchedTables);
     }
+    cacheNotify.notify(new ServerCacheNotification(modTimestamp, touchedTables));
   }
 
   /**

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

@@ -1,6 +1,7 @@
 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;
@@ -25,12 +26,13 @@ public class TransactionManagerOptions {
   final TransactionScopeManager scopeManager;
   final SpiLogManager logManager;
   final TableModState tableModState;
+  final ServerCacheNotify cacheNotify;
 
 
   public TransactionManagerOptions(boolean notifyL2CacheInForeground, ServerConfig config, TransactionScopeManager scopeManager, ClusterManager clusterManager,
                                    BackgroundExecutor backgroundExecutor, DocStoreUpdateProcessor docStoreUpdateProcessor,
                                    BeanDescriptorManager descMgr, DataSourceSupplier dataSourceSupplier, SpiProfileHandler profileHandler,
-                                   SpiLogManager logManager, TableModState tableModState) {
+                                   SpiLogManager logManager, TableModState tableModState, ServerCacheNotify cacheNotify) {
 
     this.notifyL2CacheInForeground = notifyL2CacheInForeground;
     this.config = config;
@@ -43,6 +45,7 @@ public class TransactionManagerOptions {
     this.profileHandler = profileHandler;
     this.logManager = logManager;
     this.tableModState = tableModState;
+    this.cacheNotify = cacheNotify;
   }
 
 }