Implement Iterator#remove for Cache values iter (#29633)

This commit implements the ability to remove values from a Cache using
the values iterator. This brings the values iterator in line with the
keys iterator and adds support for removing items in the cache that are
not easily found by the key used for the cache.

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

@@ -580,7 +580,8 @@ public class Cache<K, V> {
 
     /**
      * An LRU sequencing of the values in the cache. This sequence is not protected from mutations
-     * to the cache. The result of iteration under mutation is undefined.
+     * to the cache (except for {@link Iterator#remove()}. The result of iteration under any other mutation is
+     * undefined.
      *
      * @return an LRU-ordered {@link Iterable} over the values in the cache
      */
@@ -597,6 +598,11 @@ public class Cache<K, V> {
             public V next() {
                 return iterator.next().value;
             }
+
+            @Override
+            public void remove() {
+                iterator.remove();
+            }
         };
     }
 

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

@@ -27,6 +27,7 @@ import java.lang.management.ThreadMXBean;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
@@ -824,4 +825,34 @@ public class CacheTests extends ESTestCase {
         cache.refresh();
         assertEquals(500, cache.count());
     }
+
+    public void testRemoveUsingValuesIterator() {
+        final List<RemovalNotification<Integer, String>> removalNotifications = new ArrayList<>();
+        Cache<Integer, String> cache =
+            CacheBuilder.<Integer, String>builder()
+                .setMaximumWeight(numberOfEntries)
+                .removalListener(removalNotifications::add)
+                .build();
+
+        for (int i = 0; i < numberOfEntries; i++) {
+            cache.put(i, Integer.toString(i));
+        }
+
+        assertThat(removalNotifications.size(), is(0));
+        final List<String> expectedRemovals = new ArrayList<>();
+        Iterator<String> valueIterator = cache.values().iterator();
+        while (valueIterator.hasNext()) {
+            String value = valueIterator.next();
+            if (randomBoolean()) {
+                valueIterator.remove();
+                expectedRemovals.add(value);
+            }
+        }
+
+        assertEquals(expectedRemovals.size(), removalNotifications.size());
+        for (int i = 0; i < expectedRemovals.size(); i++) {
+            assertEquals(expectedRemovals.get(i), removalNotifications.get(i).getValue());
+            assertEquals(RemovalNotification.RemovalReason.INVALIDATED, removalNotifications.get(i).getRemovalReason());
+        }
+    }
 }