Reject multiple data paths on frozen capable nodes (#71896)

Nodes with a frozen cache no longer supports multiple data paths. This
simplifies cache sizing and avoids the need to support multiple cache
files.

Relates #71844

x-pack/plugin/autoscaling/src/internalClusterTest/java/org/elasticsearch/xpack/autoscaling/AbstractFrozenAutoscalingIntegTestCase.java

@@ -51,6 +51,11 @@ public abstract class AbstractFrozenAutoscalingIntegTestCase extends AbstractSna
     protected final String snapshotName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
     protected final String policyName = "frozen";
 
+    @Override
+    protected boolean forceSingleDataPath() {
+        return true;
+    }
+
     @Override
     protected boolean addMockInternalEngine() {
         return false;

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/BaseFrozenSearchableSnapshotsIntegTestCase.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.searchablesnapshots;
+
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeUnit;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.xpack.searchablesnapshots.cache.shared.FrozenCacheService;
+
+public class BaseFrozenSearchableSnapshotsIntegTestCase extends BaseSearchableSnapshotsIntegTestCase {
+    @Override
+    protected boolean forceSingleDataPath() {
+        return true;
+    }
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
+        Settings.Builder builder = Settings.builder().put(super.nodeSettings(nodeOrdinal, otherSettings));
+        if (DiscoveryNode.canContainData(otherSettings)) {
+            builder.put(
+                FrozenCacheService.SNAPSHOT_CACHE_SIZE_SETTING.getKey(),
+                rarely()
+                    ? randomBoolean()
+                        ? new ByteSizeValue(randomIntBetween(1, 10), ByteSizeUnit.KB)
+                        : new ByteSizeValue(randomIntBetween(1, 1000), ByteSizeUnit.BYTES)
+                    : new ByteSizeValue(randomIntBetween(1, 10), ByteSizeUnit.MB)
+            );
+        }
+
+        return builder.build();
+    }
+}

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/BaseSearchableSnapshotsIntegTestCase.java

@@ -14,7 +14,10 @@
  */
 package org.elasticsearch.xpack.searchablesnapshots;
 
+import org.apache.lucene.search.TotalHits;
+import org.apache.lucene.store.AlreadyClosedException;
 import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
+import org.elasticsearch.action.admin.indices.recovery.RecoveryResponse;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeRole;
@@ -22,7 +25,20 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.common.util.concurrent.AtomicArray;
+import org.elasticsearch.env.NodeEnvironment;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.engine.Engine;
+import org.elasticsearch.index.engine.EngineTestCase;
+import org.elasticsearch.index.engine.ReadOnlyEngine;
+import org.elasticsearch.index.shard.IndexShard;
+import org.elasticsearch.index.shard.IndexShardTestCase;
+import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.index.shard.ShardPath;
+import org.elasticsearch.indices.IndicesService;
+import org.elasticsearch.indices.recovery.RecoveryState;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.snapshots.AbstractSnapshotIntegTestCase;
 import org.elasticsearch.test.ESIntegTestCase;
@@ -34,18 +50,26 @@ import org.elasticsearch.xpack.searchablesnapshots.cache.full.CacheService;
 import org.elasticsearch.xpack.searchablesnapshots.cache.shared.FrozenCacheService;
 import org.junit.After;
 
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Locale;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
 
+import static org.elasticsearch.index.query.QueryBuilders.matchQuery;
 import static org.elasticsearch.license.LicenseService.SELF_GENERATED_LICENSE_TYPE;
 import static org.elasticsearch.test.NodeRoles.addRoles;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.xpack.searchablesnapshots.cache.shared.SharedBytes.pageAligned;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.instanceOf;
 
 @ESIntegTestCase.ClusterScope(supportsDedicatedMasters = false, numClientNodes = 0)
 public abstract class BaseSearchableSnapshotsIntegTestCase extends AbstractSnapshotIntegTestCase {
@@ -79,15 +103,8 @@ public abstract class BaseSearchableSnapshotsIntegTestCase extends AbstractSnaps
                     : new ByteSizeValue(randomIntBetween(1, 10), ByteSizeUnit.MB)
             );
         }
-        if (DiscoveryNode.canContainData(otherSettings)) {
-            builder.put(
-                FrozenCacheService.SNAPSHOT_CACHE_SIZE_SETTING.getKey(),
-                rarely()
-                    ? randomBoolean()
-                        ? new ByteSizeValue(randomIntBetween(1, 10), ByteSizeUnit.KB)
-                        : new ByteSizeValue(randomIntBetween(1, 1000), ByteSizeUnit.BYTES)
-                    : new ByteSizeValue(randomIntBetween(1, 10), ByteSizeUnit.MB)
-            );
+        if (DiscoveryNode.canContainData(otherSettings) && randomBoolean()) {
+            builder.put(FrozenCacheService.SNAPSHOT_CACHE_SIZE_SETTING.getKey(), ByteSizeValue.ZERO.getStringRep());
         }
         builder.put(
             FrozenCacheService.SNAPSHOT_CACHE_REGION_SIZE_SETTING.getKey(),
@@ -185,4 +202,111 @@ public abstract class BaseSearchableSnapshotsIntegTestCase extends AbstractSnaps
             );
         }
     }
+
+    protected void checkSoftDeletesNotEagerlyLoaded(String restoredIndexName) {
+        for (IndicesService indicesService : internalCluster().getDataNodeInstances(IndicesService.class)) {
+            for (IndexService indexService : indicesService) {
+                if (indexService.index().getName().equals(restoredIndexName)) {
+                    for (IndexShard indexShard : indexService) {
+                        try {
+                            Engine engine = IndexShardTestCase.getEngine(indexShard);
+                            assertThat(engine, instanceOf(ReadOnlyEngine.class));
+                            EngineTestCase.checkNoSoftDeletesLoaded((ReadOnlyEngine) engine);
+                        } catch (AlreadyClosedException ace) {
+                            // ok to ignore these
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    protected void assertShardFolders(String indexName, boolean snapshotDirectory) throws IOException {
+        final Index restoredIndex = resolveIndex(indexName);
+        final String customDataPath = resolveCustomDataPath(indexName);
+        final ShardId shardId = new ShardId(restoredIndex, 0);
+        boolean shardFolderFound = false;
+        for (String node : internalCluster().getNodeNames()) {
+            final NodeEnvironment service = internalCluster().getInstance(NodeEnvironment.class, node);
+            final ShardPath shardPath = ShardPath.loadShardPath(logger, service, shardId, customDataPath);
+            if (shardPath != null && Files.exists(shardPath.getDataPath())) {
+                shardFolderFound = true;
+                assertEquals(snapshotDirectory, Files.notExists(shardPath.resolveIndex()));
+
+                assertTrue(Files.exists(shardPath.resolveTranslog()));
+                try (Stream<Path> dir = Files.list(shardPath.resolveTranslog())) {
+                    final long translogFiles = dir.filter(path -> path.getFileName().toString().contains("translog")).count();
+                    if (snapshotDirectory) {
+                        assertEquals(2L, translogFiles);
+                    } else {
+                        assertThat(translogFiles, greaterThanOrEqualTo(2L));
+                    }
+                }
+            }
+        }
+        assertTrue("no shard folder found for index " + indexName, shardFolderFound);
+    }
+
+    protected void assertTotalHits(String indexName, TotalHits originalAllHits, TotalHits originalBarHits) throws Exception {
+        final Thread[] threads = new Thread[between(1, 5)];
+        final AtomicArray<TotalHits> allHits = new AtomicArray<>(threads.length);
+        final AtomicArray<TotalHits> barHits = new AtomicArray<>(threads.length);
+
+        final CountDownLatch latch = new CountDownLatch(1);
+        for (int i = 0; i < threads.length; i++) {
+            int t = i;
+            threads[i] = new Thread(() -> {
+                try {
+                    latch.await();
+                } catch (InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
+                allHits.set(t, client().prepareSearch(indexName).setTrackTotalHits(true).get().getHits().getTotalHits());
+                barHits.set(
+                    t,
+                    client().prepareSearch(indexName)
+                        .setTrackTotalHits(true)
+                        .setQuery(matchQuery("foo", "bar"))
+                        .get()
+                        .getHits()
+                        .getTotalHits()
+                );
+            });
+            threads[i].start();
+        }
+
+        ensureGreen(indexName);
+        latch.countDown();
+
+        for (int i = 0; i < threads.length; i++) {
+            threads[i].join();
+
+            final TotalHits allTotalHits = allHits.get(i);
+            final TotalHits barTotalHits = barHits.get(i);
+
+            logger.info("--> thread #{} has [{}] hits in total, of which [{}] match the query", i, allTotalHits, barTotalHits);
+            assertThat(allTotalHits, equalTo(originalAllHits));
+            assertThat(barTotalHits, equalTo(originalBarHits));
+        }
+    }
+
+    protected void assertRecoveryStats(String indexName, boolean preWarmEnabled) throws Exception {
+        int shardCount = getNumShards(indexName).totalNumShards;
+        assertBusy(() -> {
+            final RecoveryResponse recoveryResponse = client().admin().indices().prepareRecoveries(indexName).get();
+            assertThat(recoveryResponse.toString(), recoveryResponse.shardRecoveryStates().get(indexName).size(), equalTo(shardCount));
+
+            for (List<RecoveryState> recoveryStates : recoveryResponse.shardRecoveryStates().values()) {
+                for (RecoveryState recoveryState : recoveryStates) {
+                    RecoveryState.Index index = recoveryState.getIndex();
+                    assertThat(
+                        Strings.toString(recoveryState, true, true),
+                        index.recoveredFileCount(),
+                        preWarmEnabled ? equalTo(index.totalRecoverFiles()) : greaterThanOrEqualTo(0)
+                    );
+                    assertThat(recoveryState.getStage(), equalTo(RecoveryState.Stage.DONE));
+                }
+            }
+        }, 30L, TimeUnit.SECONDS);
+    }
 }

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/FrozenSearchableSnapshotsIntegTests.java

@@ -0,0 +1,418 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.searchablesnapshots;
+
+import org.apache.lucene.search.TotalHits;
+import org.elasticsearch.ResourceNotFoundException;
+import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
+import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotIndexShardStatus;
+import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
+import org.elasticsearch.action.admin.indices.shrink.ResizeType;
+import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
+import org.elasticsearch.action.admin.indices.stats.ShardStats;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.routing.allocation.decider.Decision;
+import org.elasticsearch.cluster.routing.allocation.decider.DiskThresholdDecider;
+import org.elasticsearch.common.Priority;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeUnit;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.index.IndexModule;
+import org.elasticsearch.index.IndexNotFoundException;
+import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.shard.IndexLongFieldRange;
+import org.elasticsearch.index.store.Store;
+import org.elasticsearch.index.store.StoreStats;
+import org.elasticsearch.indices.IndexClosedException;
+import org.elasticsearch.snapshots.SnapshotInfo;
+import org.elasticsearch.xpack.cluster.routing.allocation.DataTierAllocationDecider;
+import org.elasticsearch.xpack.core.DataTier;
+import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotAction;
+import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotRequest;
+import org.elasticsearch.xpack.searchablesnapshots.action.SearchableSnapshotsStatsAction;
+import org.elasticsearch.xpack.searchablesnapshots.action.SearchableSnapshotsStatsRequest;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.StreamSupport;
+
+import static org.elasticsearch.index.IndexSettings.INDEX_SOFT_DELETES_SETTING;
+import static org.elasticsearch.index.query.QueryBuilders.matchQuery;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshots.getDataTiersPreference;
+import static org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshotsConstants.SNAPSHOT_DIRECTORY_FACTORY_KEY;
+import static org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshotsConstants.SNAPSHOT_RECOVERY_STATE_FACTORY_KEY;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.lessThan;
+import static org.hamcrest.Matchers.oneOf;
+import static org.hamcrest.Matchers.sameInstance;
+
+public class FrozenSearchableSnapshotsIntegTests extends BaseFrozenSearchableSnapshotsIntegTestCase {
+
+    public void testCreateAndRestorePartialSearchableSnapshot() throws Exception {
+        final String fsRepoName = randomAlphaOfLength(10);
+        final String indexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        final String aliasName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        final String restoredIndexName = randomBoolean() ? indexName : randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        final String snapshotName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+
+        createRepository(
+            fsRepoName,
+            "fs",
+            Settings.builder().put("location", randomRepoPath()).put("chunk_size", randomIntBetween(100, 1000), ByteSizeUnit.BYTES)
+        );
+
+        // Peer recovery always copies .liv files but we do not permit writing to searchable snapshot directories so this doesn't work, but
+        // we can bypass this by forcing soft deletes to be used. TODO this restriction can be lifted when #55142 is resolved.
+        assertAcked(prepareCreate(indexName, Settings.builder().put(INDEX_SOFT_DELETES_SETTING.getKey(), true)));
+        assertAcked(client().admin().indices().prepareAliases().addAlias(indexName, aliasName));
+
+        populateIndex(indexName, 10_000);
+
+        final TotalHits originalAllHits = internalCluster().client()
+            .prepareSearch(indexName)
+            .setTrackTotalHits(true)
+            .get()
+            .getHits()
+            .getTotalHits();
+        final TotalHits originalBarHits = internalCluster().client()
+            .prepareSearch(indexName)
+            .setTrackTotalHits(true)
+            .setQuery(matchQuery("foo", "bar"))
+            .get()
+            .getHits()
+            .getTotalHits();
+        logger.info("--> [{}] in total, of which [{}] match the query", originalAllHits, originalBarHits);
+
+        expectThrows(
+            ResourceNotFoundException.class,
+            "Searchable snapshot stats on a non snapshot searchable index should fail",
+            () -> client().execute(SearchableSnapshotsStatsAction.INSTANCE, new SearchableSnapshotsStatsRequest()).actionGet()
+        );
+
+        final SnapshotInfo snapshotInfo = createFullSnapshot(fsRepoName, snapshotName);
+        ensureGreen(indexName);
+
+        assertShardFolders(indexName, false);
+
+        assertThat(
+            client().admin()
+                .cluster()
+                .prepareState()
+                .clear()
+                .setMetadata(true)
+                .setIndices(indexName)
+                .get()
+                .getState()
+                .metadata()
+                .index(indexName)
+                .getTimestampRange(),
+            sameInstance(IndexLongFieldRange.UNKNOWN)
+        );
+
+        final boolean deletedBeforeMount = randomBoolean();
+        if (deletedBeforeMount) {
+            assertAcked(client().admin().indices().prepareDelete(indexName));
+        } else {
+            assertAcked(client().admin().indices().prepareClose(indexName));
+        }
+
+        logger.info("--> restoring partial index [{}] with cache enabled", restoredIndexName);
+
+        Settings.Builder indexSettingsBuilder = Settings.builder()
+            .put(SearchableSnapshots.SNAPSHOT_CACHE_ENABLED_SETTING.getKey(), true)
+            .put(IndexSettings.INDEX_CHECK_ON_STARTUP.getKey(), Boolean.FALSE.toString());
+        final List<String> nonCachedExtensions;
+        if (randomBoolean()) {
+            nonCachedExtensions = randomSubsetOf(Arrays.asList("fdt", "fdx", "nvd", "dvd", "tip", "cfs", "dim"));
+            indexSettingsBuilder.putList(SearchableSnapshots.SNAPSHOT_CACHE_EXCLUDED_FILE_TYPES_SETTING.getKey(), nonCachedExtensions);
+        } else {
+            nonCachedExtensions = Collections.emptyList();
+        }
+        if (randomBoolean()) {
+            indexSettingsBuilder.put(
+                SearchableSnapshots.SNAPSHOT_UNCACHED_CHUNK_SIZE_SETTING.getKey(),
+                new ByteSizeValue(randomLongBetween(10, 100_000))
+            );
+        }
+        final int expectedReplicas;
+        if (randomBoolean()) {
+            expectedReplicas = numberOfReplicas();
+            indexSettingsBuilder.put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, expectedReplicas);
+        } else {
+            expectedReplicas = 0;
+        }
+        final String expectedDataTiersPreference;
+        expectedDataTiersPreference = getDataTiersPreference(MountSearchableSnapshotRequest.Storage.SHARED_CACHE);
+
+        indexSettingsBuilder.put(Store.INDEX_STORE_STATS_REFRESH_INTERVAL_SETTING.getKey(), TimeValue.ZERO);
+        final AtomicBoolean statsWatcherRunning = new AtomicBoolean(true);
+        final Thread statsWatcher = new Thread(() -> {
+            while (statsWatcherRunning.get()) {
+                final IndicesStatsResponse indicesStatsResponse;
+                try {
+                    indicesStatsResponse = client().admin().indices().prepareStats(restoredIndexName).clear().setStore(true).get();
+                } catch (IndexNotFoundException | IndexClosedException e) {
+                    continue;
+                    // ok
+                }
+
+                for (ShardStats shardStats : indicesStatsResponse.getShards()) {
+                    StoreStats store = shardStats.getStats().getStore();
+                    assertThat(shardStats.getShardRouting().toString(), store.getReservedSize().getBytes(), equalTo(0L));
+                    assertThat(shardStats.getShardRouting().toString(), store.getSize().getBytes(), equalTo(0L));
+                }
+                if (indicesStatsResponse.getShards().length > 0) {
+                    assertThat(indicesStatsResponse.getTotal().getStore().getReservedSize().getBytes(), equalTo(0L));
+                    assertThat(indicesStatsResponse.getTotal().getStore().getSize().getBytes(), equalTo(0L));
+                }
+            }
+        }, "test-stats-watcher");
+        statsWatcher.start();
+
+        final MountSearchableSnapshotRequest req = new MountSearchableSnapshotRequest(
+            restoredIndexName,
+            fsRepoName,
+            snapshotInfo.snapshotId().getName(),
+            indexName,
+            indexSettingsBuilder.build(),
+            Strings.EMPTY_ARRAY,
+            true,
+            MountSearchableSnapshotRequest.Storage.SHARED_CACHE
+        );
+
+        final RestoreSnapshotResponse restoreSnapshotResponse = client().execute(MountSearchableSnapshotAction.INSTANCE, req).get();
+        assertThat(restoreSnapshotResponse.getRestoreInfo().failedShards(), equalTo(0));
+
+        final Map<Integer, SnapshotIndexShardStatus> snapshotShards = clusterAdmin().prepareSnapshotStatus(fsRepoName)
+            .setSnapshots(snapshotInfo.snapshotId().getName())
+            .get()
+            .getSnapshots()
+            .get(0)
+            .getIndices()
+            .get(indexName)
+            .getShards();
+
+        ensureGreen(restoredIndexName);
+
+        final IndicesStatsResponse indicesStatsResponse = client().admin()
+            .indices()
+            .prepareStats(restoredIndexName)
+            .clear()
+            .setStore(true)
+            .get();
+        assertThat(indicesStatsResponse.getShards().length, greaterThan(0));
+        long totalExpectedSize = 0;
+        for (ShardStats shardStats : indicesStatsResponse.getShards()) {
+            StoreStats store = shardStats.getStats().getStore();
+            assertThat(shardStats.getShardRouting().toString(), store.getReservedSize().getBytes(), equalTo(0L));
+            assertThat(shardStats.getShardRouting().toString(), store.getSize().getBytes(), equalTo(0L));
+
+            // the extra segments_N file created for bootstrap new history and associate translog makes us unable to precisely assert this.
+            final long expectedSize = snapshotShards.get(shardStats.getShardRouting().getId()).getStats().getTotalSize();
+            assertThat(shardStats.getShardRouting().toString(), store.getTotalDataSetSize().getBytes(), greaterThanOrEqualTo(expectedSize));
+            // the extra segments_N file only has a new history UUID and translog UUID, both of which have constant size. It's size is
+            // therefore identical to the original segments_N file from the snapshot. We expect at least 1 byte of other content, making
+            // it safe to assert that the total data set size is less than 2x the size.
+            assertThat(shardStats.getShardRouting().toString(), store.getTotalDataSetSize().getBytes(), lessThan(expectedSize * 2));
+
+            totalExpectedSize += expectedSize;
+        }
+
+        // the extra segments_N file created for bootstrap new history and associate translog makes us unable to precisely assert this.
+        final StoreStats store = indicesStatsResponse.getTotal().getStore();
+        assertThat(store.getTotalDataSetSize().getBytes(), greaterThanOrEqualTo(totalExpectedSize));
+        assertThat(store.getTotalDataSetSize().getBytes(), lessThan(totalExpectedSize * 2));
+
+        statsWatcherRunning.set(false);
+        statsWatcher.join();
+
+        final Settings settings = client().admin()
+            .indices()
+            .prepareGetSettings(restoredIndexName)
+            .get()
+            .getIndexToSettings()
+            .get(restoredIndexName);
+        assertThat(SearchableSnapshots.SNAPSHOT_SNAPSHOT_NAME_SETTING.get(settings), equalTo(snapshotName));
+        assertThat(IndexModule.INDEX_STORE_TYPE_SETTING.get(settings), equalTo(SNAPSHOT_DIRECTORY_FACTORY_KEY));
+        assertThat(IndexModule.INDEX_RECOVERY_TYPE_SETTING.get(settings), equalTo(SNAPSHOT_RECOVERY_STATE_FACTORY_KEY));
+        assertTrue(IndexMetadata.INDEX_BLOCKS_WRITE_SETTING.get(settings));
+        assertTrue(SearchableSnapshots.SNAPSHOT_SNAPSHOT_ID_SETTING.exists(settings));
+        assertTrue(SearchableSnapshots.SNAPSHOT_INDEX_ID_SETTING.exists(settings));
+        assertThat(IndexMetadata.INDEX_AUTO_EXPAND_REPLICAS_SETTING.get(settings).toString(), equalTo("false"));
+        assertThat(IndexMetadata.INDEX_NUMBER_OF_REPLICAS_SETTING.get(settings), equalTo(expectedReplicas));
+        assertThat(DataTierAllocationDecider.INDEX_ROUTING_PREFER_SETTING.get(settings), equalTo(expectedDataTiersPreference));
+        assertTrue(SearchableSnapshotsConstants.SNAPSHOT_PARTIAL_SETTING.get(settings));
+        assertTrue(DiskThresholdDecider.SETTING_IGNORE_DISK_WATERMARKS.get(settings));
+
+        checkSoftDeletesNotEagerlyLoaded(restoredIndexName);
+        assertTotalHits(restoredIndexName, originalAllHits, originalBarHits);
+        assertRecoveryStats(restoredIndexName, false);
+        // TODO: fix
+        // assertSearchableSnapshotStats(restoredIndexName, true, nonCachedExtensions);
+        ensureGreen(restoredIndexName);
+        assertShardFolders(restoredIndexName, true);
+
+        assertThat(
+            client().admin()
+                .cluster()
+                .prepareState()
+                .clear()
+                .setMetadata(true)
+                .setIndices(restoredIndexName)
+                .get()
+                .getState()
+                .metadata()
+                .index(restoredIndexName)
+                .getTimestampRange(),
+            sameInstance(IndexLongFieldRange.UNKNOWN)
+        );
+
+        if (deletedBeforeMount) {
+            assertThat(client().admin().indices().prepareGetAliases(aliasName).get().getAliases().size(), equalTo(0));
+            assertAcked(client().admin().indices().prepareAliases().addAlias(restoredIndexName, aliasName));
+        } else if (indexName.equals(restoredIndexName) == false) {
+            assertThat(client().admin().indices().prepareGetAliases(aliasName).get().getAliases().size(), equalTo(1));
+            assertAcked(
+                client().admin()
+                    .indices()
+                    .prepareAliases()
+                    .addAliasAction(IndicesAliasesRequest.AliasActions.remove().index(indexName).alias(aliasName).mustExist(true))
+                    .addAlias(restoredIndexName, aliasName)
+            );
+        }
+        assertThat(client().admin().indices().prepareGetAliases(aliasName).get().getAliases().size(), equalTo(1));
+        assertTotalHits(aliasName, originalAllHits, originalBarHits);
+
+        final Decision diskDeciderDecision = client().admin()
+            .cluster()
+            .prepareAllocationExplain()
+            .setIndex(restoredIndexName)
+            .setShard(0)
+            .setPrimary(true)
+            .setIncludeYesDecisions(true)
+            .get()
+            .getExplanation()
+            .getShardAllocationDecision()
+            .getMoveDecision()
+            .getCanRemainDecision()
+            .getDecisions()
+            .stream()
+            .filter(d -> d.label().equals(DiskThresholdDecider.NAME))
+            .findFirst()
+            .orElseThrow();
+        assertThat(diskDeciderDecision.type(), equalTo(Decision.Type.YES));
+        assertThat(
+            diskDeciderDecision.getExplanation(),
+            oneOf("disk watermarks are ignored on this index", "there is only a single data node present")
+        );
+
+        internalCluster().fullRestart();
+        assertTotalHits(restoredIndexName, originalAllHits, originalBarHits);
+        assertRecoveryStats(restoredIndexName, false);
+        assertTotalHits(aliasName, originalAllHits, originalBarHits);
+        // TODO: fix
+        // assertSearchableSnapshotStats(restoredIndexName, false, nonCachedExtensions);
+
+        internalCluster().ensureAtLeastNumDataNodes(2);
+
+        final DiscoveryNode dataNode = randomFrom(
+            StreamSupport.stream(
+                client().admin().cluster().prepareState().get().getState().nodes().getDataNodes().values().spliterator(),
+                false
+            ).map(c -> c.value).toArray(DiscoveryNode[]::new)
+        );
+
+        assertAcked(
+            client().admin()
+                .indices()
+                .prepareUpdateSettings(restoredIndexName)
+                .setSettings(
+                    Settings.builder()
+                        .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+                        .put(
+                            IndexMetadata.INDEX_ROUTING_REQUIRE_GROUP_SETTING.getConcreteSettingForNamespace("_name").getKey(),
+                            dataNode.getName()
+                        )
+                )
+        );
+
+        assertFalse(
+            client().admin()
+                .cluster()
+                .prepareHealth(restoredIndexName)
+                .setWaitForNoRelocatingShards(true)
+                .setWaitForEvents(Priority.LANGUID)
+                .get()
+                .isTimedOut()
+        );
+
+        assertTotalHits(restoredIndexName, originalAllHits, originalBarHits);
+        assertRecoveryStats(restoredIndexName, false);
+        // TODO: fix
+        // assertSearchableSnapshotStats(restoredIndexName, false, nonCachedExtensions);
+
+        assertAcked(
+            client().admin()
+                .indices()
+                .prepareUpdateSettings(restoredIndexName)
+                .setSettings(
+                    Settings.builder()
+                        .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1)
+                        .putNull(IndexMetadata.INDEX_ROUTING_REQUIRE_GROUP_SETTING.getConcreteSettingForNamespace("_name").getKey())
+                )
+        );
+
+        assertTotalHits(restoredIndexName, originalAllHits, originalBarHits);
+        assertRecoveryStats(restoredIndexName, false);
+
+        final String clonedIndexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        assertAcked(
+            client().admin()
+                .indices()
+                .prepareResizeIndex(restoredIndexName, clonedIndexName)
+                .setResizeType(ResizeType.CLONE)
+                .setSettings(
+                    Settings.builder()
+                        .putNull(IndexModule.INDEX_STORE_TYPE_SETTING.getKey())
+                        .putNull(IndexModule.INDEX_RECOVERY_TYPE_SETTING.getKey())
+                        .put(DataTierAllocationDecider.INDEX_ROUTING_PREFER, DataTier.DATA_HOT)
+                        .build()
+                )
+        );
+        ensureGreen(clonedIndexName);
+        assertTotalHits(clonedIndexName, originalAllHits, originalBarHits);
+
+        final Settings clonedIndexSettings = client().admin()
+            .indices()
+            .prepareGetSettings(clonedIndexName)
+            .get()
+            .getIndexToSettings()
+            .get(clonedIndexName);
+        assertFalse(clonedIndexSettings.hasValue(IndexModule.INDEX_STORE_TYPE_SETTING.getKey()));
+        assertFalse(clonedIndexSettings.hasValue(SearchableSnapshots.SNAPSHOT_SNAPSHOT_NAME_SETTING.getKey()));
+        assertFalse(clonedIndexSettings.hasValue(SearchableSnapshots.SNAPSHOT_SNAPSHOT_ID_SETTING.getKey()));
+        assertFalse(clonedIndexSettings.hasValue(SearchableSnapshots.SNAPSHOT_INDEX_ID_SETTING.getKey()));
+        assertFalse(clonedIndexSettings.hasValue(IndexModule.INDEX_RECOVERY_TYPE_SETTING.getKey()));
+
+        assertAcked(client().admin().indices().prepareDelete(restoredIndexName));
+        assertThat(client().admin().indices().prepareGetAliases(aliasName).get().getAliases().size(), equalTo(0));
+        assertAcked(client().admin().indices().prepareAliases().addAlias(clonedIndexName, aliasName));
+        assertTotalHits(aliasName, originalAllHits, originalBarHits);
+    }
+
+}

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshotsCanMatchOnCoordinatorIntegTests.java

@@ -55,7 +55,7 @@ import static org.hamcrest.Matchers.nullValue;
 import static org.hamcrest.Matchers.sameInstance;
 
 @ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0)
-public class SearchableSnapshotsCanMatchOnCoordinatorIntegTests extends BaseSearchableSnapshotsIntegTestCase {
+public class SearchableSnapshotsCanMatchOnCoordinatorIntegTests extends BaseFrozenSearchableSnapshotsIntegTestCase {
 
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins() {

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshotsIntegTests.java

@@ -7,7 +7,6 @@
 package org.elasticsearch.xpack.searchablesnapshots;
 
 import org.apache.lucene.search.TotalHits;
-import org.apache.lucene.store.AlreadyClosedException;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
@@ -15,7 +14,6 @@ import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotIndexShar
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotStats;
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotStatus;
 import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
-import org.elasticsearch.action.admin.indices.recovery.RecoveryResponse;
 import org.elasticsearch.action.admin.indices.shrink.ResizeType;
 import org.elasticsearch.action.admin.indices.stats.IndexStats;
 import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
@@ -27,37 +25,19 @@ import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.ShardRouting;
-import org.elasticsearch.cluster.routing.allocation.decider.Decision;
-import org.elasticsearch.cluster.routing.allocation.decider.DiskThresholdDecider;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
-import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.common.util.concurrent.AtomicArray;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.env.Environment;
-import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexModule;
-import org.elasticsearch.index.IndexNotFoundException;
-import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.IndexSettings;
-import org.elasticsearch.index.engine.Engine;
-import org.elasticsearch.index.engine.EngineTestCase;
-import org.elasticsearch.index.engine.ReadOnlyEngine;
 import org.elasticsearch.index.mapper.DateFieldMapper;
 import org.elasticsearch.index.shard.IndexLongFieldRange;
-import org.elasticsearch.index.shard.IndexShard;
-import org.elasticsearch.index.shard.IndexShardTestCase;
-import org.elasticsearch.index.shard.ShardId;
-import org.elasticsearch.index.shard.ShardPath;
-import org.elasticsearch.index.store.Store;
-import org.elasticsearch.index.store.StoreStats;
-import org.elasticsearch.indices.IndexClosedException;
 import org.elasticsearch.indices.IndicesService;
-import org.elasticsearch.indices.recovery.RecoveryState;
 import org.elasticsearch.repositories.RepositoryData;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.snapshots.SnapshotId;
@@ -72,8 +52,6 @@ import org.elasticsearch.xpack.searchablesnapshots.action.SearchableSnapshotsSta
 import org.elasticsearch.xpack.searchablesnapshots.action.SearchableSnapshotsStatsRequest;
 import org.elasticsearch.xpack.searchablesnapshots.action.SearchableSnapshotsStatsResponse;
 
-import java.io.IOException;
-import java.nio.file.Files;
 import java.nio.file.Path;
 import java.time.Instant;
 import java.util.ArrayList;
@@ -84,13 +62,9 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.BrokenBarrierException;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CyclicBarrier;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
-import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
 import static org.elasticsearch.cluster.metadata.IndexMetadata.INDEX_NUMBER_OF_SHARDS_SETTING;
@@ -108,11 +82,8 @@ import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
-import static org.hamcrest.Matchers.instanceOf;
-import static org.hamcrest.Matchers.lessThan;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 import static org.hamcrest.Matchers.not;
-import static org.hamcrest.Matchers.oneOf;
 import static org.hamcrest.Matchers.sameInstance;
 
 public class SearchableSnapshotsIntegTests extends BaseSearchableSnapshotsIntegTestCase {
@@ -401,402 +372,6 @@ public class SearchableSnapshotsIntegTests extends BaseSearchableSnapshotsIntegT
         assertTotalHits(aliasName, originalAllHits, originalBarHits);
     }
 
-    public void testCreateAndRestorePartialSearchableSnapshot() throws Exception {
-        final String fsRepoName = randomAlphaOfLength(10);
-        final String indexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
-        final String aliasName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
-        final String restoredIndexName = randomBoolean() ? indexName : randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
-        final String snapshotName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
-
-        createRepository(
-            fsRepoName,
-            "fs",
-            Settings.builder().put("location", randomRepoPath()).put("chunk_size", randomIntBetween(100, 1000), ByteSizeUnit.BYTES)
-        );
-
-        // Peer recovery always copies .liv files but we do not permit writing to searchable snapshot directories so this doesn't work, but
-        // we can bypass this by forcing soft deletes to be used. TODO this restriction can be lifted when #55142 is resolved.
-        assertAcked(prepareCreate(indexName, Settings.builder().put(INDEX_SOFT_DELETES_SETTING.getKey(), true)));
-        assertAcked(client().admin().indices().prepareAliases().addAlias(indexName, aliasName));
-
-        populateIndex(indexName, 10_000);
-
-        final TotalHits originalAllHits = internalCluster().client()
-            .prepareSearch(indexName)
-            .setTrackTotalHits(true)
-            .get()
-            .getHits()
-            .getTotalHits();
-        final TotalHits originalBarHits = internalCluster().client()
-            .prepareSearch(indexName)
-            .setTrackTotalHits(true)
-            .setQuery(matchQuery("foo", "bar"))
-            .get()
-            .getHits()
-            .getTotalHits();
-        logger.info("--> [{}] in total, of which [{}] match the query", originalAllHits, originalBarHits);
-
-        expectThrows(
-            ResourceNotFoundException.class,
-            "Searchable snapshot stats on a non snapshot searchable index should fail",
-            () -> client().execute(SearchableSnapshotsStatsAction.INSTANCE, new SearchableSnapshotsStatsRequest()).actionGet()
-        );
-
-        final SnapshotInfo snapshotInfo = createFullSnapshot(fsRepoName, snapshotName);
-        ensureGreen(indexName);
-
-        assertShardFolders(indexName, false);
-
-        assertThat(
-            client().admin()
-                .cluster()
-                .prepareState()
-                .clear()
-                .setMetadata(true)
-                .setIndices(indexName)
-                .get()
-                .getState()
-                .metadata()
-                .index(indexName)
-                .getTimestampRange(),
-            sameInstance(IndexLongFieldRange.UNKNOWN)
-        );
-
-        final boolean deletedBeforeMount = randomBoolean();
-        if (deletedBeforeMount) {
-            assertAcked(client().admin().indices().prepareDelete(indexName));
-        } else {
-            assertAcked(client().admin().indices().prepareClose(indexName));
-        }
-
-        logger.info("--> restoring partial index [{}] with cache enabled", restoredIndexName);
-
-        Settings.Builder indexSettingsBuilder = Settings.builder()
-            .put(SearchableSnapshots.SNAPSHOT_CACHE_ENABLED_SETTING.getKey(), true)
-            .put(IndexSettings.INDEX_CHECK_ON_STARTUP.getKey(), Boolean.FALSE.toString());
-        final List<String> nonCachedExtensions;
-        if (randomBoolean()) {
-            nonCachedExtensions = randomSubsetOf(Arrays.asList("fdt", "fdx", "nvd", "dvd", "tip", "cfs", "dim"));
-            indexSettingsBuilder.putList(SearchableSnapshots.SNAPSHOT_CACHE_EXCLUDED_FILE_TYPES_SETTING.getKey(), nonCachedExtensions);
-        } else {
-            nonCachedExtensions = Collections.emptyList();
-        }
-        if (randomBoolean()) {
-            indexSettingsBuilder.put(
-                SearchableSnapshots.SNAPSHOT_UNCACHED_CHUNK_SIZE_SETTING.getKey(),
-                new ByteSizeValue(randomLongBetween(10, 100_000))
-            );
-        }
-        final int expectedReplicas;
-        if (randomBoolean()) {
-            expectedReplicas = numberOfReplicas();
-            indexSettingsBuilder.put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, expectedReplicas);
-        } else {
-            expectedReplicas = 0;
-        }
-        final String expectedDataTiersPreference;
-        expectedDataTiersPreference = getDataTiersPreference(MountSearchableSnapshotRequest.Storage.SHARED_CACHE);
-
-        indexSettingsBuilder.put(Store.INDEX_STORE_STATS_REFRESH_INTERVAL_SETTING.getKey(), TimeValue.ZERO);
-        final AtomicBoolean statsWatcherRunning = new AtomicBoolean(true);
-        final Thread statsWatcher = new Thread(() -> {
-            while (statsWatcherRunning.get()) {
-                final IndicesStatsResponse indicesStatsResponse;
-                try {
-                    indicesStatsResponse = client().admin().indices().prepareStats(restoredIndexName).clear().setStore(true).get();
-                } catch (IndexNotFoundException | IndexClosedException e) {
-                    continue;
-                    // ok
-                }
-
-                for (ShardStats shardStats : indicesStatsResponse.getShards()) {
-                    StoreStats store = shardStats.getStats().getStore();
-                    assertThat(shardStats.getShardRouting().toString(), store.getReservedSize().getBytes(), equalTo(0L));
-                    assertThat(shardStats.getShardRouting().toString(), store.getSize().getBytes(), equalTo(0L));
-                }
-                if (indicesStatsResponse.getShards().length > 0) {
-                    assertThat(indicesStatsResponse.getTotal().getStore().getReservedSize().getBytes(), equalTo(0L));
-                    assertThat(indicesStatsResponse.getTotal().getStore().getSize().getBytes(), equalTo(0L));
-                }
-            }
-        }, "test-stats-watcher");
-        statsWatcher.start();
-
-        final MountSearchableSnapshotRequest req = new MountSearchableSnapshotRequest(
-            restoredIndexName,
-            fsRepoName,
-            snapshotInfo.snapshotId().getName(),
-            indexName,
-            indexSettingsBuilder.build(),
-            Strings.EMPTY_ARRAY,
-            true,
-            MountSearchableSnapshotRequest.Storage.SHARED_CACHE
-        );
-
-        final RestoreSnapshotResponse restoreSnapshotResponse = client().execute(MountSearchableSnapshotAction.INSTANCE, req).get();
-        assertThat(restoreSnapshotResponse.getRestoreInfo().failedShards(), equalTo(0));
-
-        final Map<Integer, SnapshotIndexShardStatus> snapshotShards = clusterAdmin().prepareSnapshotStatus(fsRepoName)
-            .setSnapshots(snapshotInfo.snapshotId().getName())
-            .get()
-            .getSnapshots()
-            .get(0)
-            .getIndices()
-            .get(indexName)
-            .getShards();
-
-        ensureGreen(restoredIndexName);
-
-        final IndicesStatsResponse indicesStatsResponse = client().admin()
-            .indices()
-            .prepareStats(restoredIndexName)
-            .clear()
-            .setStore(true)
-            .get();
-        assertThat(indicesStatsResponse.getShards().length, greaterThan(0));
-        long totalExpectedSize = 0;
-        for (ShardStats shardStats : indicesStatsResponse.getShards()) {
-            StoreStats store = shardStats.getStats().getStore();
-            assertThat(shardStats.getShardRouting().toString(), store.getReservedSize().getBytes(), equalTo(0L));
-            assertThat(shardStats.getShardRouting().toString(), store.getSize().getBytes(), equalTo(0L));
-
-            // the extra segments_N file created for bootstrap new history and associate translog makes us unable to precisely assert this.
-            final long expectedSize = snapshotShards.get(shardStats.getShardRouting().getId()).getStats().getTotalSize();
-            assertThat(shardStats.getShardRouting().toString(), store.getTotalDataSetSize().getBytes(), greaterThanOrEqualTo(expectedSize));
-            // the extra segments_N file only has a new history UUID and translog UUID, both of which have constant size. It's size is
-            // therefore identical to the original segments_N file from the snapshot. We expect at least 1 byte of other content, making
-            // it safe to assert that the total data set size is less than 2x the size.
-            assertThat(shardStats.getShardRouting().toString(), store.getTotalDataSetSize().getBytes(), lessThan(expectedSize * 2));
-
-            totalExpectedSize += expectedSize;
-        }
-
-        // the extra segments_N file created for bootstrap new history and associate translog makes us unable to precisely assert this.
-        final StoreStats store = indicesStatsResponse.getTotal().getStore();
-        assertThat(store.getTotalDataSetSize().getBytes(), greaterThanOrEqualTo(totalExpectedSize));
-        assertThat(store.getTotalDataSetSize().getBytes(), lessThan(totalExpectedSize * 2));
-
-        statsWatcherRunning.set(false);
-        statsWatcher.join();
-
-        final Settings settings = client().admin()
-            .indices()
-            .prepareGetSettings(restoredIndexName)
-            .get()
-            .getIndexToSettings()
-            .get(restoredIndexName);
-        assertThat(SearchableSnapshots.SNAPSHOT_SNAPSHOT_NAME_SETTING.get(settings), equalTo(snapshotName));
-        assertThat(IndexModule.INDEX_STORE_TYPE_SETTING.get(settings), equalTo(SNAPSHOT_DIRECTORY_FACTORY_KEY));
-        assertThat(IndexModule.INDEX_RECOVERY_TYPE_SETTING.get(settings), equalTo(SNAPSHOT_RECOVERY_STATE_FACTORY_KEY));
-        assertTrue(IndexMetadata.INDEX_BLOCKS_WRITE_SETTING.get(settings));
-        assertTrue(SearchableSnapshots.SNAPSHOT_SNAPSHOT_ID_SETTING.exists(settings));
-        assertTrue(SearchableSnapshots.SNAPSHOT_INDEX_ID_SETTING.exists(settings));
-        assertThat(IndexMetadata.INDEX_AUTO_EXPAND_REPLICAS_SETTING.get(settings).toString(), equalTo("false"));
-        assertThat(IndexMetadata.INDEX_NUMBER_OF_REPLICAS_SETTING.get(settings), equalTo(expectedReplicas));
-        assertThat(DataTierAllocationDecider.INDEX_ROUTING_PREFER_SETTING.get(settings), equalTo(expectedDataTiersPreference));
-        assertTrue(SearchableSnapshotsConstants.SNAPSHOT_PARTIAL_SETTING.get(settings));
-        assertTrue(DiskThresholdDecider.SETTING_IGNORE_DISK_WATERMARKS.get(settings));
-
-        checkSoftDeletesNotEagerlyLoaded(restoredIndexName);
-        assertTotalHits(restoredIndexName, originalAllHits, originalBarHits);
-        assertRecoveryStats(restoredIndexName, false);
-        // TODO: fix
-        // assertSearchableSnapshotStats(restoredIndexName, true, nonCachedExtensions);
-        ensureGreen(restoredIndexName);
-        assertShardFolders(restoredIndexName, true);
-
-        assertThat(
-            client().admin()
-                .cluster()
-                .prepareState()
-                .clear()
-                .setMetadata(true)
-                .setIndices(restoredIndexName)
-                .get()
-                .getState()
-                .metadata()
-                .index(restoredIndexName)
-                .getTimestampRange(),
-            sameInstance(IndexLongFieldRange.UNKNOWN)
-        );
-
-        if (deletedBeforeMount) {
-            assertThat(client().admin().indices().prepareGetAliases(aliasName).get().getAliases().size(), equalTo(0));
-            assertAcked(client().admin().indices().prepareAliases().addAlias(restoredIndexName, aliasName));
-        } else if (indexName.equals(restoredIndexName) == false) {
-            assertThat(client().admin().indices().prepareGetAliases(aliasName).get().getAliases().size(), equalTo(1));
-            assertAcked(
-                client().admin()
-                    .indices()
-                    .prepareAliases()
-                    .addAliasAction(IndicesAliasesRequest.AliasActions.remove().index(indexName).alias(aliasName).mustExist(true))
-                    .addAlias(restoredIndexName, aliasName)
-            );
-        }
-        assertThat(client().admin().indices().prepareGetAliases(aliasName).get().getAliases().size(), equalTo(1));
-        assertTotalHits(aliasName, originalAllHits, originalBarHits);
-
-        final Decision diskDeciderDecision = client().admin()
-            .cluster()
-            .prepareAllocationExplain()
-            .setIndex(restoredIndexName)
-            .setShard(0)
-            .setPrimary(true)
-            .setIncludeYesDecisions(true)
-            .get()
-            .getExplanation()
-            .getShardAllocationDecision()
-            .getMoveDecision()
-            .getCanRemainDecision()
-            .getDecisions()
-            .stream()
-            .filter(d -> d.label().equals(DiskThresholdDecider.NAME))
-            .findFirst()
-            .orElseThrow();
-        assertThat(diskDeciderDecision.type(), equalTo(Decision.Type.YES));
-        assertThat(
-            diskDeciderDecision.getExplanation(),
-            oneOf("disk watermarks are ignored on this index", "there is only a single data node present")
-        );
-
-        internalCluster().fullRestart();
-        assertTotalHits(restoredIndexName, originalAllHits, originalBarHits);
-        assertRecoveryStats(restoredIndexName, false);
-        assertTotalHits(aliasName, originalAllHits, originalBarHits);
-        // TODO: fix
-        // assertSearchableSnapshotStats(restoredIndexName, false, nonCachedExtensions);
-
-        internalCluster().ensureAtLeastNumDataNodes(2);
-
-        final DiscoveryNode dataNode = randomFrom(
-            StreamSupport.stream(
-                client().admin().cluster().prepareState().get().getState().nodes().getDataNodes().values().spliterator(),
-                false
-            ).map(c -> c.value).toArray(DiscoveryNode[]::new)
-        );
-
-        assertAcked(
-            client().admin()
-                .indices()
-                .prepareUpdateSettings(restoredIndexName)
-                .setSettings(
-                    Settings.builder()
-                        .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
-                        .put(
-                            IndexMetadata.INDEX_ROUTING_REQUIRE_GROUP_SETTING.getConcreteSettingForNamespace("_name").getKey(),
-                            dataNode.getName()
-                        )
-                )
-        );
-
-        assertFalse(
-            client().admin()
-                .cluster()
-                .prepareHealth(restoredIndexName)
-                .setWaitForNoRelocatingShards(true)
-                .setWaitForEvents(Priority.LANGUID)
-                .get()
-                .isTimedOut()
-        );
-
-        assertTotalHits(restoredIndexName, originalAllHits, originalBarHits);
-        assertRecoveryStats(restoredIndexName, false);
-        // TODO: fix
-        // assertSearchableSnapshotStats(restoredIndexName, false, nonCachedExtensions);
-
-        assertAcked(
-            client().admin()
-                .indices()
-                .prepareUpdateSettings(restoredIndexName)
-                .setSettings(
-                    Settings.builder()
-                        .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1)
-                        .putNull(IndexMetadata.INDEX_ROUTING_REQUIRE_GROUP_SETTING.getConcreteSettingForNamespace("_name").getKey())
-                )
-        );
-
-        assertTotalHits(restoredIndexName, originalAllHits, originalBarHits);
-        assertRecoveryStats(restoredIndexName, false);
-
-        final String clonedIndexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
-        assertAcked(
-            client().admin()
-                .indices()
-                .prepareResizeIndex(restoredIndexName, clonedIndexName)
-                .setResizeType(ResizeType.CLONE)
-                .setSettings(
-                    Settings.builder()
-                        .putNull(IndexModule.INDEX_STORE_TYPE_SETTING.getKey())
-                        .putNull(IndexModule.INDEX_RECOVERY_TYPE_SETTING.getKey())
-                        .put(DataTierAllocationDecider.INDEX_ROUTING_PREFER, DataTier.DATA_HOT)
-                        .build()
-                )
-        );
-        ensureGreen(clonedIndexName);
-        assertTotalHits(clonedIndexName, originalAllHits, originalBarHits);
-
-        final Settings clonedIndexSettings = client().admin()
-            .indices()
-            .prepareGetSettings(clonedIndexName)
-            .get()
-            .getIndexToSettings()
-            .get(clonedIndexName);
-        assertFalse(clonedIndexSettings.hasValue(IndexModule.INDEX_STORE_TYPE_SETTING.getKey()));
-        assertFalse(clonedIndexSettings.hasValue(SearchableSnapshots.SNAPSHOT_SNAPSHOT_NAME_SETTING.getKey()));
-        assertFalse(clonedIndexSettings.hasValue(SearchableSnapshots.SNAPSHOT_SNAPSHOT_ID_SETTING.getKey()));
-        assertFalse(clonedIndexSettings.hasValue(SearchableSnapshots.SNAPSHOT_INDEX_ID_SETTING.getKey()));
-        assertFalse(clonedIndexSettings.hasValue(IndexModule.INDEX_RECOVERY_TYPE_SETTING.getKey()));
-
-        assertAcked(client().admin().indices().prepareDelete(restoredIndexName));
-        assertThat(client().admin().indices().prepareGetAliases(aliasName).get().getAliases().size(), equalTo(0));
-        assertAcked(client().admin().indices().prepareAliases().addAlias(clonedIndexName, aliasName));
-        assertTotalHits(aliasName, originalAllHits, originalBarHits);
-    }
-
-    private void checkSoftDeletesNotEagerlyLoaded(String restoredIndexName) {
-        for (IndicesService indicesService : internalCluster().getDataNodeInstances(IndicesService.class)) {
-            for (IndexService indexService : indicesService) {
-                if (indexService.index().getName().equals(restoredIndexName)) {
-                    for (IndexShard indexShard : indexService) {
-                        try {
-                            Engine engine = IndexShardTestCase.getEngine(indexShard);
-                            assertThat(engine, instanceOf(ReadOnlyEngine.class));
-                            EngineTestCase.checkNoSoftDeletesLoaded((ReadOnlyEngine) engine);
-                        } catch (AlreadyClosedException ace) {
-                            // ok to ignore these
-                        }
-                    }
-                }
-            }
-        }
-    }
-
-    private void assertShardFolders(String indexName, boolean snapshotDirectory) throws IOException {
-        final Index restoredIndex = resolveIndex(indexName);
-        final String customDataPath = resolveCustomDataPath(indexName);
-        final ShardId shardId = new ShardId(restoredIndex, 0);
-        boolean shardFolderFound = false;
-        for (String node : internalCluster().getNodeNames()) {
-            final NodeEnvironment service = internalCluster().getInstance(NodeEnvironment.class, node);
-            final ShardPath shardPath = ShardPath.loadShardPath(logger, service, shardId, customDataPath);
-            if (shardPath != null && Files.exists(shardPath.getDataPath())) {
-                shardFolderFound = true;
-                assertEquals(snapshotDirectory, Files.notExists(shardPath.resolveIndex()));
-
-                assertTrue(Files.exists(shardPath.resolveTranslog()));
-                try (Stream<Path> dir = Files.list(shardPath.resolveTranslog())) {
-                    final long translogFiles = dir.filter(path -> path.getFileName().toString().contains("translog")).count();
-                    if (snapshotDirectory) {
-                        assertEquals(2L, translogFiles);
-                    } else {
-                        assertThat(translogFiles, greaterThanOrEqualTo(2L));
-                    }
-                }
-            }
-        }
-        assertTrue("no shard folder found", shardFolderFound);
-    }
-
     public void testCanMountSnapshotTakenWhileConcurrentlyIndexing() throws Exception {
         final String fsRepoName = randomAlphaOfLength(10);
         final String indexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
@@ -1392,69 +967,6 @@ public class SearchableSnapshotsIntegTests extends BaseSearchableSnapshotsIntegT
         );
     }
 
-    private void assertTotalHits(String indexName, TotalHits originalAllHits, TotalHits originalBarHits) throws Exception {
-        final Thread[] threads = new Thread[between(1, 5)];
-        final AtomicArray<TotalHits> allHits = new AtomicArray<>(threads.length);
-        final AtomicArray<TotalHits> barHits = new AtomicArray<>(threads.length);
-
-        final CountDownLatch latch = new CountDownLatch(1);
-        for (int i = 0; i < threads.length; i++) {
-            int t = i;
-            threads[i] = new Thread(() -> {
-                try {
-                    latch.await();
-                } catch (InterruptedException e) {
-                    throw new RuntimeException(e);
-                }
-                allHits.set(t, client().prepareSearch(indexName).setTrackTotalHits(true).get().getHits().getTotalHits());
-                barHits.set(
-                    t,
-                    client().prepareSearch(indexName)
-                        .setTrackTotalHits(true)
-                        .setQuery(matchQuery("foo", "bar"))
-                        .get()
-                        .getHits()
-                        .getTotalHits()
-                );
-            });
-            threads[i].start();
-        }
-
-        ensureGreen(indexName);
-        latch.countDown();
-
-        for (int i = 0; i < threads.length; i++) {
-            threads[i].join();
-
-            final TotalHits allTotalHits = allHits.get(i);
-            final TotalHits barTotalHits = barHits.get(i);
-
-            logger.info("--> thread #{} has [{}] hits in total, of which [{}] match the query", i, allTotalHits, barTotalHits);
-            assertThat(allTotalHits, equalTo(originalAllHits));
-            assertThat(barTotalHits, equalTo(originalBarHits));
-        }
-    }
-
-    private void assertRecoveryStats(String indexName, boolean preWarmEnabled) throws Exception {
-        int shardCount = getNumShards(indexName).totalNumShards;
-        assertBusy(() -> {
-            final RecoveryResponse recoveryResponse = client().admin().indices().prepareRecoveries(indexName).get();
-            assertThat(recoveryResponse.toString(), recoveryResponse.shardRecoveryStates().get(indexName).size(), equalTo(shardCount));
-
-            for (List<RecoveryState> recoveryStates : recoveryResponse.shardRecoveryStates().values()) {
-                for (RecoveryState recoveryState : recoveryStates) {
-                    RecoveryState.Index index = recoveryState.getIndex();
-                    assertThat(
-                        Strings.toString(recoveryState, true, true),
-                        index.recoveredFileCount(),
-                        preWarmEnabled ? equalTo(index.totalRecoverFiles()) : greaterThanOrEqualTo(0)
-                    );
-                    assertThat(recoveryState.getStage(), equalTo(RecoveryState.Stage.DONE));
-                }
-            }
-        }, 30L, TimeUnit.SECONDS);
-    }
-
     private void assertSearchableSnapshotStats(String indexName, boolean cacheEnabled, List<String> nonCachedExtensions) {
         final SearchableSnapshotsStatsResponse statsResponse = client().execute(
             SearchableSnapshotsStatsAction.INSTANCE,

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshotsLicenseIntegTests.java

@@ -55,7 +55,7 @@ import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.notNullValue;
 
 @ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST)
-public class SearchableSnapshotsLicenseIntegTests extends BaseSearchableSnapshotsIntegTestCase {
+public class SearchableSnapshotsLicenseIntegTests extends BaseFrozenSearchableSnapshotsIntegTestCase {
 
     private static final String repoName = "test-repo";
     private static final String indexName = "test-index";

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshotsSettingValidationIntegTests.java

@@ -22,7 +22,7 @@ import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 
 @ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST)
-public class SearchableSnapshotsSettingValidationIntegTests extends BaseSearchableSnapshotsIntegTestCase {
+public class SearchableSnapshotsSettingValidationIntegTests extends BaseFrozenSearchableSnapshotsIntegTestCase {
 
     private static final String repoName = "test-repo";
     private static final String indexName = "test-index";

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshotsShardLimitIntegTests.java

@@ -19,7 +19,7 @@ import static org.elasticsearch.index.IndexSettings.INDEX_SOFT_DELETES_SETTING;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 
 @ESIntegTestCase.ClusterScope(maxNumDataNodes = 1)
-public class SearchableSnapshotsShardLimitIntegTests extends BaseSearchableSnapshotsIntegTestCase {
+public class SearchableSnapshotsShardLimitIntegTests extends BaseFrozenSearchableSnapshotsIntegTestCase {
 
     private static final int MAX_NORMAL = 3;
     private static final int MAX_FROZEN = 20;

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshotsSystemIndicesIntegTests.java

@@ -31,7 +31,7 @@ import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcke
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 
-public class SearchableSnapshotsSystemIndicesIntegTests extends BaseSearchableSnapshotsIntegTestCase {
+public class SearchableSnapshotsSystemIndicesIntegTests extends BaseFrozenSearchableSnapshotsIntegTestCase {
 
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins() {

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshotsUuidValidationIntegTests.java

@@ -35,7 +35,7 @@ import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcke
 import static org.hamcrest.Matchers.containsString;
 
 @ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST)
-public class SearchableSnapshotsUuidValidationIntegTests extends BaseSearchableSnapshotsIntegTestCase {
+public class SearchableSnapshotsUuidValidationIntegTests extends BaseFrozenSearchableSnapshotsIntegTestCase {
 
     public static class TestPlugin extends Plugin implements ActionPlugin {
 

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/allocation/SearchableSnapshotDataTierIntegTests.java

@@ -13,12 +13,12 @@ import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.xpack.cluster.routing.allocation.DataTierAllocationDecider;
 import org.elasticsearch.xpack.core.DataTier;
 import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotRequest;
-import org.elasticsearch.xpack.searchablesnapshots.BaseSearchableSnapshotsIntegTestCase;
+import org.elasticsearch.xpack.searchablesnapshots.BaseFrozenSearchableSnapshotsIntegTestCase;
 
 import java.util.Map;
 
 @ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST)
-public class SearchableSnapshotDataTierIntegTests extends BaseSearchableSnapshotsIntegTestCase {
+public class SearchableSnapshotDataTierIntegTests extends BaseFrozenSearchableSnapshotsIntegTestCase {
 
     private static final String repoName = "test-repo";
     private static final String indexName = "test-index";

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/cache/blob/SearchableSnapshotsBlobStoreCacheIntegTests.java

@@ -39,7 +39,7 @@ import org.elasticsearch.xpack.cluster.routing.allocation.DataTierAllocationDeci
 import org.elasticsearch.xpack.core.ClientHelper;
 import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotRequest.Storage;
 import org.elasticsearch.xpack.core.searchablesnapshots.SearchableSnapshotShardStats;
-import org.elasticsearch.xpack.searchablesnapshots.BaseSearchableSnapshotsIntegTestCase;
+import org.elasticsearch.xpack.searchablesnapshots.BaseFrozenSearchableSnapshotsIntegTestCase;
 import org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshots;
 import org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshotsConstants;
 import org.elasticsearch.xpack.searchablesnapshots.action.SearchableSnapshotsStatsAction;
@@ -63,7 +63,7 @@ import static org.elasticsearch.xpack.searchablesnapshots.cache.shared.SharedByt
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 
-public class SearchableSnapshotsBlobStoreCacheIntegTests extends BaseSearchableSnapshotsIntegTestCase {
+public class SearchableSnapshotsBlobStoreCacheIntegTests extends BaseFrozenSearchableSnapshotsIntegTestCase {
 
     private static Settings cacheSettings = null;
     private static ByteSizeValue blobCacheMaxLength = null;

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

@@ -26,6 +26,7 @@ import org.elasticsearch.common.util.concurrent.AbstractAsyncTask;
 import org.elasticsearch.common.util.concurrent.AbstractRefCounted;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.util.concurrent.KeyedLock;
+import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.node.NodeRoleSettings;
@@ -117,12 +118,23 @@ public class FrozenCacheService implements Releasable {
                             roles.stream().map(DiscoveryNodeRole::roleName).collect(Collectors.joining(","))
                         );
                     }
+
+                    @SuppressWarnings("unchecked")
+                    final List<String> dataPaths = (List<String>) settings.get(Environment.PATH_DATA_SETTING);
+                    if (dataPaths.size() > 1) {
+                        throw new SettingsException(
+                            "setting [{}={}] is not permitted on nodes with multiple data paths [{}]",
+                            SNAPSHOT_CACHE_SIZE_SETTING.getKey(),
+                            value.getStringRep(),
+                            String.join(",", dataPaths)
+                        );
+                    }
                 }
             }
 
             @Override
             public Iterator<Setting<?>> settings() {
-                final List<Setting<?>> settings = List.of(NodeRoleSettings.NODE_ROLES_SETTING);
+                final List<Setting<?>> settings = List.of(NodeRoleSettings.NODE_ROLES_SETTING, Environment.PATH_DATA_SETTING);
                 return settings.iterator();
             }
 

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

@@ -9,7 +9,6 @@ package org.elasticsearch.xpack.searchablesnapshots.cache.shared;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.SuppressForbidden;
 import org.elasticsearch.common.io.Channels;
 import org.elasticsearch.common.unit.ByteSizeValue;
@@ -62,9 +61,6 @@ public class SharedBytes extends AbstractRefCounted {
         Path cacheFile = null;
         if (fileSize > 0) {
             cacheFile = findCacheSnapshotCacheFilePath(environment, fileSize);
-            if (cacheFile == null) {
-                throw new IOException("Could not find a directory with adequate free space for cache file");
-            }
             Preallocate.preallocate(cacheFile, fileSize);
             // TODO: maybe make this faster by allocating a larger direct buffer if this is too slow for very large files
             // We fill either the full file or the bytes between its current size and the desired size once with zeros to fully allocate
@@ -103,24 +99,22 @@ public class SharedBytes extends AbstractRefCounted {
      *
      * @return path for the cache file or {@code null} if none could be found
      */
-    @Nullable
     public static Path findCacheSnapshotCacheFilePath(NodeEnvironment environment, long fileSize) throws IOException {
-        Path cacheFile = null;
-        for (Path path : environment.nodeDataPaths()) {
-            Files.createDirectories(path);
-            // TODO: be resilient to this check failing and try next path?
-            long usableSpace = Environment.getUsableSpace(path);
-            Path p = path.resolve(CACHE_FILE_NAME);
-            if (Files.exists(p)) {
-                usableSpace += Files.size(p);
-            }
-            // TODO: leave some margin for error here
-            if (usableSpace > fileSize) {
-                cacheFile = p;
-                break;
-            }
+        assert environment.nodeDataPaths().length == 1;
+        Path path = environment.nodeDataPaths()[0];
+        Files.createDirectories(path);
+        // TODO: be resilient to this check failing and try next path?
+        long usableSpace = Environment.getUsableSpace(path);
+        Path p = path.resolve(CACHE_FILE_NAME);
+        if (Files.exists(p)) {
+            usableSpace += Files.size(p);
+        }
+        // TODO: leave some margin for error here
+        if (usableSpace > fileSize) {
+            return p;
+        } else {
+            throw new IOException("Not enough free space for cache file of size [" + fileSize + "] in path [" + path + "]");
         }
-        return cacheFile;
     }
 
     @Override

x-pack/plugin/searchable-snapshots/src/test/java/org/elasticsearch/xpack/searchablesnapshots/AbstractSearchableSnapshotsTestCase.java

@@ -31,7 +31,9 @@ import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.core.internal.io.IOUtils;
+import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.NodeEnvironment;
+import org.elasticsearch.env.TestEnvironment;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.store.Store;
 import org.elasticsearch.indices.recovery.RecoveryState;
@@ -88,6 +90,7 @@ public abstract class AbstractSearchableSnapshotsTestCase extends ESIndexInputTe
     protected ThreadPool threadPool;
     protected ClusterService clusterService;
     protected NodeEnvironment nodeEnvironment;
+    protected NodeEnvironment singlePathNodeEnvironment;
 
     @Before
     public void setUpTest() throws Exception {
@@ -101,11 +104,13 @@ public abstract class AbstractSearchableSnapshotsTestCase extends ESIndexInputTe
         threadPool = new TestThreadPool(getTestName(), SearchableSnapshots.executorBuilders(Settings.EMPTY));
         clusterService = ClusterServiceUtils.createClusterService(threadPool, node, CLUSTER_SETTINGS);
         nodeEnvironment = newNodeEnvironment();
+        singlePathNodeEnvironment = newSinglePathNodeEnvironment();
     }
 
     @After
     public void tearDownTest() throws Exception {
         IOUtils.close(nodeEnvironment, clusterService);
+        IOUtils.close(singlePathNodeEnvironment, clusterService);
         assertTrue(ThreadPool.terminate(threadPool, 30L, TimeUnit.SECONDS));
     }
 
@@ -157,7 +162,7 @@ public abstract class AbstractSearchableSnapshotsTestCase extends ESIndexInputTe
         if (randomBoolean()) {
             cacheSettings.put(FrozenCacheService.FROZEN_CACHE_RECOVERY_RANGE_SIZE_SETTING.getKey(), randomFrozenCacheRangeSize());
         }
-        return new FrozenCacheService(nodeEnvironment, cacheSettings.build(), threadPool);
+        return new FrozenCacheService(singlePathNodeEnvironment, cacheSettings.build(), threadPool);
     }
 
     /**
@@ -174,7 +179,7 @@ public abstract class AbstractSearchableSnapshotsTestCase extends ESIndexInputTe
 
     protected FrozenCacheService createFrozenCacheService(final ByteSizeValue cacheSize, final ByteSizeValue cacheRangeSize) {
         return new FrozenCacheService(
-            nodeEnvironment,
+            singlePathNodeEnvironment,
             Settings.builder()
                 .put(FrozenCacheService.SNAPSHOT_CACHE_SIZE_SETTING.getKey(), cacheSize)
                 .put(FrozenCacheService.SHARED_CACHE_RANGE_SIZE_SETTING.getKey(), cacheRangeSize)
@@ -183,6 +188,14 @@ public abstract class AbstractSearchableSnapshotsTestCase extends ESIndexInputTe
         );
     }
 
+    private NodeEnvironment newSinglePathNodeEnvironment() throws IOException {
+        Settings build = Settings.builder()
+            .put(buildEnvSettings(Settings.EMPTY))
+            .putList(Environment.PATH_DATA_SETTING.getKey(), createTempDir().toAbsolutePath().toString())
+            .build();
+        return new NodeEnvironment(build, TestEnvironment.newEnvironment(build));
+    }
+
     /**
      * Returns a random shard data path for the specified {@link ShardId}. The returned path can be located on any of the data node paths.
      */

x-pack/plugin/searchable-snapshots/src/test/java/org/elasticsearch/xpack/searchablesnapshots/cache/shared/FrozenCacheServiceTests.java

@@ -12,6 +12,7 @@ import org.elasticsearch.cluster.node.DiscoveryNodeRole;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsException;
 import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.env.TestEnvironment;
 import org.elasticsearch.index.shard.ShardId;
@@ -23,6 +24,7 @@ import org.elasticsearch.xpack.searchablesnapshots.cache.common.CacheKey;
 import org.elasticsearch.xpack.searchablesnapshots.cache.shared.FrozenCacheService.CacheFileRegion;
 
 import java.io.IOException;
+import java.util.List;
 
 import static org.elasticsearch.node.Node.NODE_NAME_SETTING;
 import static org.hamcrest.Matchers.instanceOf;
@@ -222,6 +224,30 @@ public class FrozenCacheServiceTests extends ESTestCase {
         );
     }
 
+    public void testMultipleDataPathsRejectedOnFrozenNodes() {
+        final Settings settings = Settings.builder()
+            .put(FrozenCacheService.SNAPSHOT_CACHE_SIZE_SETTING.getKey(), new ByteSizeValue(size(500)).getStringRep())
+            .putList(NodeRoleSettings.NODE_ROLES_SETTING.getKey(), DiscoveryNodeRole.DATA_FROZEN_NODE_ROLE.roleName())
+            .putList(Environment.PATH_DATA_SETTING.getKey(), List.of("a", "b"))
+            .build();
+        final IllegalArgumentException e = expectThrows(
+            IllegalArgumentException.class,
+            () -> FrozenCacheService.SNAPSHOT_CACHE_SIZE_SETTING.get(settings)
+        );
+        assertThat(e.getCause(), notNullValue());
+        assertThat(e.getCause(), instanceOf(SettingsException.class));
+        assertThat(
+            e.getCause().getMessage(),
+            is(
+                "setting ["
+                    + FrozenCacheService.SNAPSHOT_CACHE_SIZE_SETTING.getKey()
+                    + "="
+                    + new ByteSizeValue(size(500)).getStringRep()
+                    + "] is not permitted on nodes with multiple data paths [a,b]"
+            )
+        );
+    }
+
     private static CacheKey generateCacheKey() {
         return new CacheKey(
             randomAlphaOfLength(10),