Fix cache expire after access

core/src/main/java/org/elasticsearch/common/cache/Cache.java

@@ -34,6 +34,7 @@
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.BiFunction;
+import java.util.function.Predicate;
 import java.util.function.ToLongBiFunction;
 
 /**
@@ -195,31 +196,32 @@ private static class CacheSegment<K, V> {
         /**
          * get an entry from the segment
          *
-         * @param key the key of the entry to get from the cache
-         * @param now the access time of this entry
+         * @param key       the key of the entry to get from the cache
+         * @param now       the access time of this entry
+         * @param isExpired test if the entry is expired
          * @return the entry if there was one, otherwise null
          */
-        Entry<K, V> get(K key, long now) {
+        Entry<K, V> get(K key, long now, Predicate<Entry<K, V>> isExpired) {
             CompletableFuture<Entry<K, V>> future;
             Entry<K, V> entry = null;
             try (ReleasableLock ignored = readLock.acquire()) {
                 future = map.get(key);
             }
             if (future != null) {
-              try {
-                  entry = future.handle((ok, ex) -> {
-                      if (ok != null) {
-                          segmentStats.hit();
-                          ok.accessTime = now;
-                          return ok;
-                      } else {
-                          segmentStats.miss();
-                          return null;
-                      }
-                  }).get();
-              } catch (ExecutionException | InterruptedException e) {
-                  throw new IllegalStateException(e);
-              }
+                try {
+                    entry = future.handle((ok, ex) -> {
+                        if (ok != null && !isExpired.test(ok)) {
+                            segmentStats.hit();
+                            ok.accessTime = now;
+                            return ok;
+                        } else {
+                            segmentStats.miss();
+                            return null;
+                        }
+                    }).get();
+                } catch (ExecutionException | InterruptedException e) {
+                    throw new IllegalStateException(e);
+                }
             }
             else {
                 segmentStats.miss();
@@ -332,8 +334,8 @@ public V get(K key) {
 
     private V get(K key, long now) {
         CacheSegment<K, V> segment = getCacheSegment(key);
-        Entry<K, V> entry = segment.get(key, now);
-        if (entry == null || isExpired(entry, now)) {
+        Entry<K, V> entry = segment.get(key, now, e -> isExpired(e, now));
+        if (entry == null) {
             return null;
         } else {
             promote(entry, now);

core/src/test/java/org/elasticsearch/common/cache/CacheTests.java

@@ -257,6 +257,28 @@ protected long now() {
         }
     }
 
+    public void testSimpleExpireAfterAccess() {
+        AtomicLong now = new AtomicLong();
+        Cache<Integer, String> cache = new Cache<Integer, String>() {
+            @Override
+            protected long now() {
+                return now.get();
+            }
+        };
+        cache.setExpireAfterAccessNanos(1);
+        now.set(0);
+        for (int i = 0; i < numberOfEntries; i++) {
+            cache.put(i, Integer.toString(i));
+        }
+        for (int i = 0; i < numberOfEntries; i++) {
+            assertEquals(cache.get(i), Integer.toString(i));
+        }
+        now.set(2);
+        for(int i = 0; i < numberOfEntries; i++) {
+            assertNull(cache.get(i));
+        }
+    }
+
     public void testExpirationAfterWrite() {
         AtomicLong now = new AtomicLong();
         Cache<Integer, String> cache = new Cache<Integer, String>() {