Wait for shard index folder to be cleaned up (#80341)

The tests testCreateAndRestoreSearchableSnapshot and 
testCreateAndRestorePartialSearchableSnapshot both 
failed once when asserting the shard folders using 
assertShardFolders(index, true).

The failures occurred when the original index is first closed 
(not deleted) and mounted again under the same name (so 
it will be restored as a searchable snapshot index on top of 
the existing shard files). The SearchableSnapshotDirectory 
implementation takes care to clean up the shard files on disk 
using SearchableSnapshotDirectory.cleanExistingRegularShardFiles() 
and the tests verify that the shard index folder is indeed deleted 
from disk on all nodes but sometime fail because the folder 
is still present.

I wasn't able to reproduce but I think that the closing of the 
original index + the creation of the .snapshot-blob-cache 
index trigger some shard relocations that are cancelled by 
the subsequent mount/restore, leaving some files on disk 
that should be cleaned up but maybe not immediately.

This commit changes the tests to assertBusy() when 
verifying the shard folders and also adds more logging 
information in case waiting for the 
assertShardFolders(index, true) is not enough.

Closes #77831

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

@@ -18,6 +18,7 @@ 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.CollectionUtils;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.index.Index;
@@ -65,6 +66,7 @@ import static org.elasticsearch.xpack.searchablesnapshots.cache.shared.SharedByt
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.not;
 
 @ESIntegTestCase.ClusterScope(supportsDedicatedMasters = false, numClientNodes = 0)
 public abstract class BaseSearchableSnapshotsIntegTestCase extends AbstractSnapshotIntegTestCase {
@@ -75,7 +77,7 @@ public abstract class BaseSearchableSnapshotsIntegTestCase extends AbstractSnaps
 
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins() {
-        return List.of(LocalStateSearchableSnapshots.class);
+        return CollectionUtils.appendToCopy(super.nodePlugins(), LocalStateSearchableSnapshots.class);
     }
 
     @Override
@@ -234,9 +236,17 @@ public abstract class BaseSearchableSnapshotsIntegTestCase extends AbstractSnaps
             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()));
+                final boolean indexExists = Files.exists(shardPath.resolveIndex());
+                final boolean translogExists = Files.exists(shardPath.resolveTranslog());
+                logger.info(
+                    "--> [{}] verifying shard data path [{}] (index exists: {}, translog exists: {})",
+                    node,
+                    shardPath.getDataPath(),
+                    indexExists,
+                    translogExists
+                );
+                assertThat(snapshotDirectory, not(indexExists));
+                assertTrue(translogExists);
                 try (Stream<Path> dir = Files.list(shardPath.resolveTranslog())) {
                     final long translogFiles = dir.filter(path -> path.getFileName().toString().contains("translog")).count();
                     if (snapshotDirectory) {

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

@@ -309,7 +309,7 @@ public class FrozenSearchableSnapshotsIntegTests extends BaseFrozenSearchableSna
         // TODO: fix
         // assertSearchableSnapshotStats(restoredIndexName, true, nonCachedExtensions);
         ensureGreen(restoredIndexName);
-        assertShardFolders(restoredIndexName, true);
+        assertBusy(() -> assertShardFolders(restoredIndexName, true));
 
         assertThat(
             client().admin()

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

@@ -254,7 +254,7 @@ public class SearchableSnapshotsIntegTests extends BaseSearchableSnapshotsIntegT
         assertSearchableSnapshotStats(restoredIndexName, cacheEnabled, nonCachedExtensions);
 
         ensureGreen(restoredIndexName);
-        assertShardFolders(restoredIndexName, true);
+        assertBusy(() -> assertShardFolders(restoredIndexName, true));
 
         assertThat(
             client().admin()