Revert #122047 (#122227) (#122321)

(cherry picked from commit 3ec1d12)

Author: mhl-b

docs/changelog/122047.yaml

@@ -508,12 +508,6 @@ public void taskSucceeded(ClusterStateTaskListener clusterStateTaskListener, Obj
             .orElseThrow()
             .queue();
 
-        // There is one task in the queue for computing and forking the cleanup work.
-        assertThat(queueLength.getAsInt(), equalTo(1));
-
-        safeAwait(barrier); // unblock the barrier thread and let it process the queue
-        safeAwait(barrier); // wait for the queue to be processed
-
         // There are indexCount (=3*snapshotPoolSize) index-deletion tasks, plus one for cleaning up the root metadata. However, the
         // throttled runner only enqueues one task per SNAPSHOT thread to start with, and then the eager runner adds another one. This shows
         // we are not spamming the threadpool with all the tasks at once, which means that other snapshot activities can run alongside this

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

@@ -29,7 +29,6 @@
 import org.elasticsearch.snapshots.SnapshotInfo;
 import org.elasticsearch.snapshots.SnapshotState;
 import org.elasticsearch.snapshots.SnapshotsService;
-import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
 
@@ -378,7 +377,6 @@ private static boolean isIndexToUpdateAfterRemovingSnapshots(
      * @return map of index to index metadata blob id to delete
      */
     public Map<IndexId, Collection<String>> indexMetaDataToRemoveAfterRemovingSnapshots(Collection<SnapshotId> snapshotIds) {
-        assert ThreadPool.assertCurrentThreadPool(ThreadPool.Names.SNAPSHOT);
         Iterator<IndexId> indicesForSnapshot = indicesToUpdateAfterRemovingSnapshot(snapshotIds);
         final Set<String> allRemainingIdentifiers = indexMetaDataGenerations.lookup.entrySet()
             .stream()

server/src/main/java/org/elasticsearch/repositories/RepositoryData.java

@@ -1134,20 +1134,14 @@ private void runWithUniqueShardMetadataNaming(ActionListener<RepositoryData> rep
                     );
                 })
 
-                .<RepositoryData>andThen(
-                    // writeIndexGen finishes on master-service thread so must fork here.
-                    snapshotExecutor,
-                    threadPool.getThreadContext(),
-                    (l, newRepositoryData) -> {
-                        l.onResponse(newRepositoryData);
-                        // Once we have updated the repository, run the unreferenced blobs cleanup in parallel to shard-level snapshot
-                        // deletion
-                        try (var refs = new RefCountingRunnable(onCompletion)) {
-                            cleanupUnlinkedRootAndIndicesBlobs(newRepositoryData, refs.acquireListener());
-                            cleanupUnlinkedShardLevelBlobs(refs.acquireListener());
-                        }
+                .<RepositoryData>andThen((l, newRepositoryData) -> {
+                    l.onResponse(newRepositoryData);
+                    // Once we have updated the repository, run the unreferenced blobs cleanup in parallel to shard-level snapshot deletion
+                    try (var refs = new RefCountingRunnable(onCompletion)) {
+                        cleanupUnlinkedRootAndIndicesBlobs(newRepositoryData, refs.acquireListener());
+                        cleanupUnlinkedShardLevelBlobs(refs.acquireListener());
                     }
-                )
+                })
 
                 .addListener(repositoryDataUpdateListener);
         }