Cleanup Snapshot ITs Further (#63985)

Drying up a few more spots and extracting tests that use their own custom plugins to separate
classes to make the giant `DedicatedClusterSnapshotRestoreIT` suite a little more digestible.

server/src/internalClusterTest/java/org/elasticsearch/snapshots/AbortedRestoreIT.java

@@ -24,16 +24,13 @@
 import org.elasticsearch.action.admin.indices.recovery.RecoveryResponse;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.cluster.routing.RecoverySource;
-import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.indices.recovery.RecoveryState;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.threadpool.ThreadPoolStats;
 import org.hamcrest.Matcher;
 
 import java.util.List;
 import java.util.concurrent.TimeUnit;
-import java.util.stream.StreamSupport;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.hamcrest.Matchers.equalTo;
@@ -85,7 +82,8 @@ public void testAbortedRestoreAlsoAbortFileRestores() throws Exception {
             }
         });
 
-        final ThreadPool.Info snapshotThreadPoolInfo = threadPool(dataNode).info(ThreadPool.Names.SNAPSHOT);
+        final ThreadPool.Info snapshotThreadPoolInfo =
+                internalCluster().getInstance(ThreadPool.class, dataNode).info(ThreadPool.Names.SNAPSHOT);
         assertThat(snapshotThreadPoolInfo.getMax(), greaterThan(0));
 
         logger.info("--> waiting for snapshot thread [max={}] pool to be full", snapshotThreadPoolInfo.getMax());
@@ -107,17 +105,6 @@ public void testAbortedRestoreAlsoAbortFileRestores() throws Exception {
     }
 
     private static void waitForMaxActiveSnapshotThreads(final String node, final Matcher<Integer> matcher) throws Exception {
-        assertBusy(() -> assertThat(threadPoolStats(node, ThreadPool.Names.SNAPSHOT).getActive(), matcher), 30L, TimeUnit.SECONDS);
-    }
-
-    private static ThreadPool threadPool(final String node) {
-        return internalCluster().getInstance(ClusterService.class, node).getClusterApplierService().threadPool();
-    }
-
-    private static ThreadPoolStats.Stats threadPoolStats(final String node, final String threadPoolName) {
-        return StreamSupport.stream(threadPool(node).stats().spliterator(), false)
-            .filter(threadPool -> threadPool.getName().equals(threadPoolName))
-            .findFirst()
-            .orElseThrow(() -> new AssertionError("Failed to find thread pool " + threadPoolName));
+        assertBusy(() -> assertThat(snapshotThreadPoolStats(node).getActive(), matcher), 30L, TimeUnit.SECONDS);
     }
 }

server/src/internalClusterTest/java/org/elasticsearch/snapshots/BlobStoreIncrementalityIT.java

@@ -36,6 +36,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.concurrent.ExecutionException;
 
 import static org.hamcrest.Matchers.greaterThan;
@@ -78,16 +79,15 @@ public void testIncrementalBehaviorOnPrimaryFailover() throws InterruptedExcepti
         final String repo = "test-repo";
         createRepository(repo, "fs");
 
-        logger.info("--> creating snapshot 1");
-        client().admin().cluster().prepareCreateSnapshot(repo, snapshot1).setIndices(indexName).setWaitForCompletion(true).get();
+        createSnapshot(repo, snapshot1, Collections.singletonList(indexName));
 
         logger.info("--> Shutting down initial primary node [{}]", primaryNode);
         stopNode(primaryNode);
 
         ensureYellow(indexName);
         final String snapshot2 = "snap-2";
         logger.info("--> creating snapshot 2");
-        client().admin().cluster().prepareCreateSnapshot(repo, snapshot2).setIndices(indexName).setWaitForCompletion(true).get();
+        createSnapshot(repo, snapshot2, Collections.singletonList(indexName));
 
         assertTwoIdenticalShardSnapshots(repo, indexName, snapshot1, snapshot2);