Be more specific when clearing cache for a specified shard (#66011)

Today we evict all cache files that belong to a given searchable
snapshot directory when this directory is closed or if the Clear
Cache API is used. This eviction iterates over all cache entries
and find the entries that matches a given shard. We can be
 more specific than that by recreating the cache keys to evict
from the list of file names instead of iterating over entries.

Backport of  #66003

Author: tlrx

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/index/store/SearchableSnapshotDirectory.java

@@ -337,7 +337,9 @@ public final void close() {
     }
 
     public void clearCache() {
-        cacheService.removeFromCache(cacheKey -> cacheKey.belongsTo(snapshotId, indexId, shardId));
+        for (BlobStoreIndexShardSnapshot.FileInfo file : files()) {
+            cacheService.removeFromCache(createCacheKey(file.physicalName()));
+        }
     }
 
     protected IndexInputStats createIndexInputStats(final long fileLength) {

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/index/store/cache/CacheKey.java

@@ -65,10 +65,4 @@ public int hashCode() {
     public String toString() {
         return "[" + "snapshotId=" + snapshotId + ", indexId=" + indexId + ", shardId=" + shardId + ", fileName='" + fileName + "']";
     }
-
-    public boolean belongsTo(SnapshotId snapshotId, IndexId indexId, ShardId shardId) {
-        return Objects.equals(this.snapshotId, snapshotId)
-            && Objects.equals(this.indexId, indexId)
-            && Objects.equals(this.shardId, shardId);
-    }
 }

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/cache/CacheService.java

@@ -22,7 +22,6 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Objects;
-import java.util.function.Predicate;
 
 import static org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshotsConstants.toIntBytes;
 
@@ -129,16 +128,11 @@ public CacheFile get(final CacheKey cacheKey, final long fileLength, final Path
     }
 
     /**
-     * Invalidate cache entries with keys matching the given predicate
+     * Evicts the cache file associated with the specified cache key.
      *
-     * @param predicate the predicate to evaluate
+     * @param cacheKey the {@link CacheKey} whose associated {@link CacheFile} must be evicted from cache
      */
-    public void removeFromCache(final Predicate<CacheKey> predicate) {
-        for (CacheKey cacheKey : cache.keys()) {
-            if (predicate.test(cacheKey)) {
-                cache.invalidate(cacheKey);
-            }
-        }
-        cache.refresh();
+    public void removeFromCache(final CacheKey cacheKey) {
+        cache.invalidate(cacheKey);
     }
 }

x-pack/plugin/searchable-snapshots/src/test/java/org/elasticsearch/index/store/cache/CacheKeyTests.java

@@ -12,45 +12,12 @@
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.EqualsHashCodeTestUtils;
 
-import static org.hamcrest.Matchers.equalTo;
-
 public class CacheKeyTests extends ESTestCase {
 
     public void testEqualsAndHashCode() {
         EqualsHashCodeTestUtils.checkEqualsAndHashCode(createInstance(), this::copy, this::mutate);
     }
 
-    public void testBelongsTo() {
-        final CacheKey cacheKey = createInstance();
-
-        SnapshotId snapshotId = cacheKey.getSnapshotId();
-        IndexId indexId = cacheKey.getIndexId();
-        ShardId shardId = cacheKey.getShardId();
-
-        final boolean belongsTo;
-        switch (randomInt(2)) {
-            case 0:
-                snapshotId = randomValueOtherThan(cacheKey.getSnapshotId(), this::randomSnapshotId);
-                belongsTo = false;
-                break;
-            case 1:
-                indexId = randomValueOtherThan(cacheKey.getIndexId(), this::randomIndexId);
-                belongsTo = false;
-                break;
-            case 2:
-                shardId = randomValueOtherThan(cacheKey.getShardId(), this::randomShardId);
-                belongsTo = false;
-                break;
-            case 3:
-                belongsTo = true;
-                break;
-            default:
-                throw new AssertionError("Unsupported value");
-        }
-
-        assertThat(cacheKey.belongsTo(snapshotId, indexId, shardId), equalTo(belongsTo));
-    }
-
     private SnapshotId randomSnapshotId() {
         return new SnapshotId(randomAlphaOfLengthBetween(5, 10), randomAlphaOfLengthBetween(5, 10));
     }