ISPN-6405 Persistence configuration with clustered cache can cause

duplicate expiration messages

* Fixed issue where reaper caused 2 expirations in cluster cache

core/src/main/java/org/infinispan/commands/write/RemoveExpiredCommand.java

@@ -118,8 +118,8 @@ public byte getCommandId() {
    public String toString() {
       return "RemoveExpiredCommand{" +
               "key=" + key +
-              "value=" + value +
-              "lifespan=" + lifespan +
+              ", value=" + value +
+              ", lifespan=" + lifespan +
               '}';
    }
 

core/src/main/java/org/infinispan/expiration/impl/ClusterExpirationManager.java

@@ -69,7 +69,7 @@ public void processExpiration() {
                InternalCacheEntry<K, V> e = purgeCandidates.next();
                if (e.canExpire()) {
                   if (ExpiryHelper.isExpiredMortal(e.getLifespan(), e.getCreated(), currentTimeMillis)) {
-                     handleLifespanExpireEntry(e);
+                     handleLifespanExpireEntry(e, true);
                   } else if (ExpiryHelper.isExpiredTransient(e.getMaxIdle(), e.getLastUsed(), currentTimeMillis)) {
                      super.handleInMemoryExpiration(e, currentTimeMillis);
                   }
@@ -89,21 +89,26 @@ public void processExpiration() {
       }
    }
 
-   void handleLifespanExpireEntry(InternalCacheEntry<K, V> entry) {
+   void handleLifespanExpireEntry(InternalCacheEntry<K, V> entry, boolean sync) {
       K key = entry.getKey();
       // The most used case will be a miss so no extra read before
       if (expiring.putIfAbsent(key, key) == null) {
          long lifespan = entry.getLifespan();
          if (trace) {
             log.tracef("Submitting expiration removal for key %s which had lifespan of %s", key, lifespan);
          }
-         asyncExecutor.submit(() -> {
+         Runnable runnable = () -> {
             try {
                removeExpired(key, entry.getValue(), lifespan);
             } finally {
                expiring.remove(key);
             }
-         });
+         };
+         if (sync) {
+            runnable.run();
+         } else {
+            asyncExecutor.submit(runnable);
+         }
       }
    }
 
@@ -117,7 +122,7 @@ public void handleInMemoryExpiration(InternalCacheEntry<K, V> entry, long curren
       // so we can see both the new value and the metadata
       synchronized (entry) {
          if (ExpiryHelper.isExpiredMortal(entry.getLifespan(), entry.getCreated(), currentTime)) {
-            handleLifespanExpireEntry(entry);
+            handleLifespanExpireEntry(entry, false);
          } else {
             super.handleInMemoryExpiration(entry, currentTime);
          }
@@ -126,25 +131,27 @@ public void handleInMemoryExpiration(InternalCacheEntry<K, V> entry, long curren
 
    @Override
    public void handleInStoreExpiration(K key) {
-      expiring.put(key, key);
-      // Unfortunately stores don't pull the entry so we can't tell exactly why it expired and thus we have to remove
-      // the entire value.  Unfortunately this could cause a concurrent write to be undone
-      try {
-         removeExpired(key, null, null);
-      } finally {
-         expiring.remove(key);
+      if (expiring.putIfAbsent(key, key) == null) {
+         // Unfortunately stores don't pull the entry so we can't tell exactly why it expired and thus we have to remove
+         // the entire value.  Unfortunately this could cause a concurrent write to be undone
+         try {
+            removeExpired(key, null, null);
+         } finally {
+            expiring.remove(key);
+         }
       }
    }
 
    @Override
    public void handleInStoreExpiration(MarshalledEntry<K, V> marshalledEntry) {
       K key = marshalledEntry.getKey();
-      expiring.put(key, key);
-      try {
-         InternalMetadata metadata = marshalledEntry.getMetadata();
-         removeExpired(key, marshalledEntry.getValue(), metadata.lifespan() == -1 ? null : metadata.lifespan());
-      } finally {
-         expiring.remove(key);
+      if (expiring.putIfAbsent(key, key) == null) {
+         try {
+            InternalMetadata metadata = marshalledEntry.getMetadata();
+            removeExpired(key, marshalledEntry.getValue(), metadata.lifespan() == -1 ? null : metadata.lifespan());
+         } finally {
+            expiring.remove(key);
+         }
       }
    }
 }

core/src/main/java/org/infinispan/persistence/file/SingleFileStore.java

@@ -659,48 +659,42 @@ private void mergeFreeEntries(List<FileEntry> entries) {
 
    @Override
    public void purge(Executor threadPool, final PurgeListener task) {
-
-      threadPool.execute(new Runnable() {
-         @Override
-         public void run() {
-            long now = timeService.wallClockTime();
-            List<KeyValuePair<Object, FileEntry>> entriesToPurge = new ArrayList<KeyValuePair<Object, FileEntry>>();
-            synchronized (entries) {
-               for (Iterator<Map.Entry<K, FileEntry>> it = entries.entrySet().iterator(); it.hasNext(); ) {
-                  Map.Entry<K, FileEntry> next = it.next();
-                  FileEntry fe = next.getValue();
-                  if (fe.isExpired(now)) {
-                     it.remove();
-                     entriesToPurge.add(new KeyValuePair<Object, FileEntry>(next.getKey(), fe));
-                  }
-               }
+      long now = timeService.wallClockTime();
+      List<KeyValuePair<Object, FileEntry>> entriesToPurge = new ArrayList<KeyValuePair<Object, FileEntry>>();
+      synchronized (entries) {
+         for (Iterator<Map.Entry<K, FileEntry>> it = entries.entrySet().iterator(); it.hasNext(); ) {
+            Map.Entry<K, FileEntry> next = it.next();
+            FileEntry fe = next.getValue();
+            if (fe.isExpired(now)) {
+               it.remove();
+               entriesToPurge.add(new KeyValuePair<Object, FileEntry>(next.getKey(), fe));
             }
+         }
+      }
 
-            resizeLock.readLock().lock();
-            try {
-               for (Iterator<KeyValuePair<Object, FileEntry>> it = entriesToPurge.iterator(); it.hasNext(); ) {
-                  KeyValuePair<Object, FileEntry> next = it.next();
-                  FileEntry fe = next.getValue();
-                  if (fe.isExpired(now)) {
-                     it.remove();
-                     try {
-                        free(fe);
-                     } catch (Exception e) {
-                        throw new PersistenceException(e);
-                     }
-                     if (task != null) task.entryPurged(next.getKey());
-                  }
-               }
-
-               // Disk space optimizations
-               synchronized (freeList) {
-                 processFreeEntries();
+      resizeLock.readLock().lock();
+      try {
+         for (Iterator<KeyValuePair<Object, FileEntry>> it = entriesToPurge.iterator(); it.hasNext(); ) {
+            KeyValuePair<Object, FileEntry> next = it.next();
+            FileEntry fe = next.getValue();
+            if (fe.isExpired(now)) {
+               it.remove();
+               try {
+                  free(fe);
+               } catch (Exception e) {
+                  throw new PersistenceException(e);
                }
-            } finally {
-               resizeLock.readLock().unlock();
+               if (task != null) task.entryPurged(next.getKey());
             }
          }
-      });
+
+         // Disk space optimizations
+         synchronized (freeList) {
+           processFreeEntries();
+         }
+      } finally {
+         resizeLock.readLock().unlock();
+      }
    }
 
    @Override

core/src/main/java/org/infinispan/persistence/spi/AdvancedCacheWriter.java

@@ -23,7 +23,10 @@ public interface AdvancedCacheWriter<K, V> extends CacheWriter<K, V> {
    /**
     * Using the thread in the pool, removed all the expired data from the persistence storage. For each removed entry,
     * the supplied listener is invoked.
-    *
+    * <p>
+    * When this method returns all entries will be purged and no tasks will be running due to this loader in the
+    * provided executor.  If however an exception is thrown there could be tasks still pending or running in the
+    * executor.
     * @throws PersistenceException in case of an error, e.g. communicating with the external storage
     */
    void purge(Executor threadPool, PurgeListener<? super K> listener);

core/src/test/java/org/infinispan/expiration/impl/ExpirationFunctionalTest.java

@@ -22,15 +22,19 @@ public class ExpirationFunctionalTest extends SingleCacheManagerTest {
    protected EmbeddedCacheManager createCacheManager() throws Exception {
       ConfigurationBuilder builder = TestCacheManagerFactory.getDefaultCacheConfiguration(false);
       configure(builder);
-      EmbeddedCacheManager cm = TestCacheManagerFactory.createCacheManager(builder);
+      EmbeddedCacheManager cm = createCacheManager(builder);
       TestingUtil.replaceComponent(cm, TimeService.class, timeService, true);
       cache = cm.getCache();
       afterCacheCreated(cm);
       return cm;
    }
 
-   protected void configure(ConfigurationBuilder config) {
+   protected EmbeddedCacheManager createCacheManager(ConfigurationBuilder builder) {
+      return TestCacheManagerFactory.createCacheManager(builder);
+   }
 
+   protected void configure(ConfigurationBuilder config) {
+      config.expiration().disableReaper();
    }
 
    protected void afterCacheCreated(EmbeddedCacheManager cm) {

core/src/test/java/org/infinispan/expiration/impl/ExpirationSingleFileStoreDistListenerFunctionalTest.java

@@ -0,0 +1,24 @@
+package org.infinispan.expiration.impl;
+
+import org.infinispan.configuration.cache.CacheMode;
+import org.infinispan.configuration.cache.ConfigurationBuilder;
+import org.infinispan.manager.EmbeddedCacheManager;
+import org.infinispan.test.fwk.TestCacheManagerFactory;
+import org.testng.annotations.Test;
+
+@Test(groups = "functional", testName = "expiration.impl.ExpirationSingleFileStoreDistListenerFunctionalTest")
+public class ExpirationSingleFileStoreDistListenerFunctionalTest extends ExpirationStoreListenerFunctionalTest {
+   @Override
+   protected void configure(ConfigurationBuilder config) {
+      config
+              // Prevent the reaper from running, reaperEnabled(false) doesn't work when a store is present
+              .expiration().wakeUpInterval(Long.MAX_VALUE)
+              .clustering().cacheMode(CacheMode.DIST_SYNC)
+              .persistence().addSingleFileStore();
+   }
+
+   @Override
+   protected EmbeddedCacheManager createCacheManager(ConfigurationBuilder builder) {
+      return TestCacheManagerFactory.createClusteredCacheManager(builder);
+   }
+}

core/src/test/java/org/infinispan/expiration/impl/ExpirationSingleFileStoreListenerFunctionalTest.java

@@ -0,0 +1,17 @@
+package org.infinispan.expiration.impl;
+
+import org.infinispan.configuration.cache.ConfigurationBuilder;
+import org.infinispan.persistence.dummy.DummyInMemoryStoreConfigurationBuilder;
+import org.infinispan.persistence.file.SingleFileStore;
+import org.testng.annotations.Test;
+
+@Test(groups = "functional", testName = "expiration.impl.ExpirationSingleFileStoreListenerFunctionalTest")
+public class ExpirationSingleFileStoreListenerFunctionalTest extends ExpirationStoreListenerFunctionalTest {
+   @Override
+   protected void configure(ConfigurationBuilder config) {
+      config
+              // Prevent the reaper from running, reaperEnabled(false) doesn't work when a store is present
+              .expiration().wakeUpInterval(Long.MAX_VALUE)
+              .persistence().addSingleFileStore();
+   }
+}

core/src/test/java/org/infinispan/expiration/impl/ExpirationStoreFunctionalTest.java

@@ -9,6 +9,9 @@ public class ExpirationStoreFunctionalTest extends ExpirationFunctionalTest {
 
    @Override
    protected void configure(ConfigurationBuilder config) {
-      config.persistence().addStore(DummyInMemoryStoreConfigurationBuilder.class);
+      config
+              // Prevent the reaper from running, reaperEnabled(false) doesn't work when a store is present
+              .expiration().wakeUpInterval(Long.MAX_VALUE)
+              .persistence().addStore(DummyInMemoryStoreConfigurationBuilder.class);
    }
 }

core/src/test/java/org/infinispan/expiration/impl/ExpirationStoreListenerFunctionalTest.java

@@ -4,6 +4,7 @@
 import org.infinispan.manager.EmbeddedCacheManager;
 import org.infinispan.notifications.cachelistener.event.CacheEntryExpiredEvent;
 import org.infinispan.notifications.cachelistener.event.Event;
+import org.infinispan.persistence.spi.AdvancedCacheExpirationWriter;
 import org.infinispan.test.TestingUtil;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.Test;
@@ -56,18 +57,20 @@ public void testExpirationOfStoreWhenDataNotInMemory() throws Exception {
 
       assertEquals(1, listener.getInvocationCount());
 
-      CacheEntryExpiredEvent event = listener.events.iterator().next();
+      CacheEntryExpiredEvent event = listener.getEvents().iterator().next();
       assertEquals(Event.Type.CACHE_ENTRY_EXPIRED, event.getType());
       assertEquals(cache, event.getCache());
       assertFalse(event.isPre());
       assertNotNull(event.getKey());
       // The dummy store produces value and metadata so lets make sure
-      assertEquals("v", event.getValue());
-      assertNotNull(event.getMetadata());
+      if (TestingUtil.getCacheLoader(cache) instanceof AdvancedCacheExpirationWriter) {
+         assertEquals("v", event.getValue());
+         assertNotNull(event.getMetadata());
+      }
    }
 
    private void assertExpiredEvents(int count) {
-      assertEquals(count, listener.getInvocationCount());
+      eventuallyEquals(count, () -> listener.getInvocationCount());
       listener.getEvents().forEach(event -> {
          assertEquals(Event.Type.CACHE_ENTRY_EXPIRED, event.getType());
          assertEquals(cache, event.getCache());

core/src/test/java/org/infinispan/expiration/impl/ExpiredCacheListener.java

@@ -3,35 +3,41 @@
 import org.infinispan.notifications.Listener;
 import org.infinispan.notifications.cachelistener.annotation.CacheEntryExpired;
 import org.infinispan.notifications.cachelistener.event.CacheEntryExpiredEvent;
+import org.infinispan.util.logging.Log;
+import org.infinispan.util.logging.LogFactory;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
 
 @Listener
 public class ExpiredCacheListener {
-   List<CacheEntryExpiredEvent> events = new ArrayList<>();
-   int invocationCount;
+   private final static Log log = LogFactory.getLog(ExpiredCacheListener.class);
+   private final List<CacheEntryExpiredEvent> events = Collections.synchronizedList(new ArrayList<>());
+   private final AtomicInteger invocationCount = new AtomicInteger();
 
    public void reset() {
       events.clear();
-      invocationCount = 0;
+      invocationCount.set(0);
    }
 
    public List<CacheEntryExpiredEvent> getEvents() {
       return events;
    }
 
    public int getInvocationCount() {
-      return invocationCount;
+      return invocationCount.get();
    }
 
 
    // handler
 
    @CacheEntryExpired
    public void handle(CacheEntryExpiredEvent e) {
+      log.trace("Received event: " + e);
       events.add(e);
 
-      invocationCount++;
+      invocationCount.incrementAndGet();
    }
 }