Simplify some Common ActionRunnable Uses (#47799)

Especially in the snapshot code there's a lot
of logic chaining `ActionRunnables` in tricky
ways now and the code is getting hard to follow.
This change introduces two convinience methods that
make it clear that a wrapped listener is invoked with
certainty in some trickier spots and shortens the code a bit.

Author: original-brownbear

plugins/repository-azure/src/main/java/org/elasticsearch/repositories/azure/AzureBlobContainer.java

@@ -147,13 +147,7 @@ public void deleteBlobsIgnoringIfNotExists(List<String> blobNames) throws IOExce
             // Executing deletes in parallel since Azure SDK 8 is using blocking IO while Azure does not provide a bulk delete API endpoint
             // TODO: Upgrade to newer non-blocking Azure SDK 11 and execute delete requests in parallel that way.
             for (String blobName : blobNames) {
-                executor.execute(new ActionRunnable<>(listener) {
-                    @Override
-                    protected void doRun() throws IOException {
-                        deleteBlobIgnoringIfNotExists(blobName);
-                        listener.onResponse(null);
-                    }
-                });
+                executor.execute(ActionRunnable.run(listener, () -> deleteBlobIgnoringIfNotExists(blobName)));
             }
         }
         try {

server/src/main/java/org/elasticsearch/action/ActionRunnable.java

@@ -20,6 +20,8 @@
 package org.elasticsearch.action;
 
 import org.elasticsearch.common.CheckedConsumer;
+import org.elasticsearch.common.CheckedRunnable;
+import org.elasticsearch.common.CheckedSupplier;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 
 /**
@@ -30,6 +32,32 @@ public abstract class ActionRunnable<Response> extends AbstractRunnable {
 
     protected final ActionListener<Response> listener;
 
+    /**
+     * Creates a {@link Runnable} that invokes the given listener with {@code null} after the given runnable has executed.
+     * @param listener Listener to invoke
+     * @param runnable Runnable to execute
+     * @return Wrapped {@code Runnable}
+     */
+    public static <T> ActionRunnable<T> run(ActionListener<T> listener, CheckedRunnable<Exception> runnable) {
+        return new ActionRunnable<>(listener) {
+            @Override
+            protected void doRun() throws Exception {
+                runnable.run();
+                listener.onResponse(null);
+            }
+        };
+    }
+
+    /**
+     * Creates a {@link Runnable} that invokes the given listener with the return of the given supplier.
+     * @param listener Listener to invoke
+     * @param supplier Supplier that provides value to pass to listener
+     * @return Wrapped {@code Runnable}
+     */
+    public static <T> ActionRunnable<T> supply(ActionListener<T> listener, CheckedSupplier<T, Exception> supplier) {
+        return ActionRunnable.wrap(listener, l -> l.onResponse(supplier.get()));
+    }
+
     /**
      * Creates a {@link Runnable} that wraps the given listener and a consumer of it that is executed when the {@link Runnable} is run.
      * Invokes {@link ActionListener#onFailure(Exception)} on it if an exception is thrown on executing the consumer.

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/get/TransportGetSnapshotsAction.java

@@ -123,17 +123,14 @@ private void getMultipleReposSnapshotInfo(List<RepositoryMetaData> repos, String
 
         // run concurrently for all repos on GENERIC thread pool
         for (final RepositoryMetaData repo : repos) {
-            threadPool.executor(ThreadPool.Names.GENERIC).execute(new ActionRunnable<>(groupedActionListener) {
-                @Override
-                protected void doRun() {
-                    try {
-                        groupedActionListener.onResponse(GetSnapshotsResponse.Response.snapshots(
-                                repo.name(), getSingleRepoSnapshotInfo(repo.name(), snapshots, ignoreUnavailable, verbose)));
-                    } catch (ElasticsearchException e) {
-                        groupedActionListener.onResponse(GetSnapshotsResponse.Response.error(repo.name(), e));
-                    }
+            threadPool.executor(ThreadPool.Names.GENERIC).execute(ActionRunnable.supply(groupedActionListener, () -> {
+                try {
+                    return GetSnapshotsResponse.Response.snapshots(
+                        repo.name(), getSingleRepoSnapshotInfo(repo.name(), snapshots, ignoreUnavailable, verbose));
+                } catch (ElasticsearchException e) {
+                    return GetSnapshotsResponse.Response.error(repo.name(), e);
                 }
-            });
+            }));
         }
     }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportSnapshotsStatusAction.java

@@ -126,8 +126,8 @@ protected void masterOperation(Task task, final SnapshotsStatusRequest request,
                     .snapshots(snapshots).timeout(request.masterNodeTimeout()),
                 ActionListener.wrap(
                     nodeSnapshotStatuses -> threadPool.executor(ThreadPool.Names.GENERIC).execute(
-                        ActionRunnable.wrap(listener, l -> l.onResponse(buildResponse(request, snapshotsService.currentSnapshots(
-                            request.repository(), Arrays.asList(request.snapshots())), nodeSnapshotStatuses)))), listener::onFailure));
+                        ActionRunnable.supply(listener, () -> buildResponse(request, snapshotsService.currentSnapshots(
+                            request.repository(), Arrays.asList(request.snapshots())), nodeSnapshotStatuses))), listener::onFailure));
         } else {
             // We don't have any in-progress shards, just return current stats
             listener.onResponse(buildResponse(request, currentSnapshots, null));

server/src/main/java/org/elasticsearch/action/support/broadcast/TransportBroadcastAction.java

@@ -300,6 +300,6 @@ public void messageReceived(ShardRequest request, TransportChannel channel, Task
 
     private void asyncShardOperation(ShardRequest request, Task task, ActionListener<ShardResponse> listener) {
         transportService.getThreadPool().executor(shardExecutor)
-            .execute(ActionRunnable.wrap(listener, l -> l.onResponse(shardOperation(request, task))));
+            .execute(ActionRunnable.supply(listener, () -> shardOperation(request, task)));
     }
 }

server/src/main/java/org/elasticsearch/action/support/single/shard/TransportSingleShardAction.java

@@ -107,7 +107,7 @@ protected void doExecute(Task task, Request request, ActionListener<Response> li
 
     protected void asyncShardOperation(Request request, ShardId shardId, ActionListener<Response> listener) throws IOException {
         threadPool.executor(getExecutor(request, shardId))
-            .execute(ActionRunnable.wrap(listener, l -> l.onResponse((shardOperation(request, shardId)))));
+            .execute(ActionRunnable.supply(listener, () -> shardOperation(request, shardId)));
     }
 
     protected abstract Writeable.Reader<Response> getResponseReader();

server/src/main/java/org/elasticsearch/indices/recovery/RecoverySourceHandler.java

@@ -406,10 +406,7 @@ private Releasable acquireStore(Store store) {
             // TODO: We shouldn't use the generic thread pool here as we already execute this from the generic pool.
             //       While practically unlikely at a min pool size of 128 we could technically block the whole pool by waiting on futures
             //       below and thus make it impossible for the store release to execute which in turn would block the futures forever
-            threadPool.generic().execute(ActionRunnable.wrap(future, l -> {
-                store.decRef();
-                l.onResponse(null);
-            }));
+            threadPool.generic().execute(ActionRunnable.run(future, store::decRef));
             FutureUtils.get(future);
         });
     }

server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java

@@ -448,14 +448,13 @@ private void cleanupStaleBlobs(Map<String, BlobContainer> foundIndices, Map<Stri
         }, listener::onFailure), 2);
 
         final Executor executor = threadPool.executor(ThreadPool.Names.SNAPSHOT);
-        executor.execute(ActionRunnable.wrap(groupedListener, l -> {
+        executor.execute(ActionRunnable.supply(groupedListener, () -> {
             List<String> deletedBlobs = cleanupStaleRootFiles(staleRootBlobs(newRepoData, rootBlobs.keySet()));
-            l.onResponse(
-                new DeleteResult(deletedBlobs.size(), deletedBlobs.stream().mapToLong(name -> rootBlobs.get(name).length()).sum()));
+            return new DeleteResult(deletedBlobs.size(), deletedBlobs.stream().mapToLong(name -> rootBlobs.get(name).length()).sum());
         }));
 
         final Set<String> survivingIndexIds = newRepoData.getIndices().values().stream().map(IndexId::getId).collect(Collectors.toSet());
-        executor.execute(ActionRunnable.wrap(groupedListener, l -> l.onResponse(cleanupStaleIndices(foundIndices, survivingIndexIds))));
+        executor.execute(ActionRunnable.supply(groupedListener, () -> cleanupStaleIndices(foundIndices, survivingIndexIds)));
     }
 
     /**
@@ -670,26 +669,22 @@ public void finalizeSnapshot(final SnapshotId snapshotId,
         // that decrements the generation it points at
 
         // Write Global MetaData
-        executor.execute(ActionRunnable.wrap(allMetaListener, l -> {
-            globalMetaDataFormat.write(clusterMetaData, blobContainer(), snapshotId.getUUID(), false);
-            l.onResponse(null);
-        }));
+        executor.execute(ActionRunnable.run(allMetaListener,
+            () -> globalMetaDataFormat.write(clusterMetaData, blobContainer(), snapshotId.getUUID(), false)));
 
         // write the index metadata for each index in the snapshot
         for (IndexId index : indices) {
-            executor.execute(ActionRunnable.wrap(allMetaListener, l -> {
-                indexMetaDataFormat.write(clusterMetaData.index(index.getName()), indexContainer(index), snapshotId.getUUID(), false);
-                l.onResponse(null);
-            }));
+            executor.execute(ActionRunnable.run(allMetaListener, () ->
+                indexMetaDataFormat.write(clusterMetaData.index(index.getName()), indexContainer(index), snapshotId.getUUID(), false)));
         }
 
-        executor.execute(ActionRunnable.wrap(allMetaListener, afterMetaListener -> {
+        executor.execute(ActionRunnable.supply(allMetaListener, () -> {
             final SnapshotInfo snapshotInfo = new SnapshotInfo(snapshotId,
                 indices.stream().map(IndexId::getName).collect(Collectors.toList()),
                 startTime, failure, threadPool.absoluteTimeInMillis(), totalShards, shardFailures,
                 includeGlobalState, userMetadata);
             snapshotFormat.write(snapshotInfo, blobContainer(), snapshotId.getUUID(), false);
-            afterMetaListener.onResponse(snapshotInfo);
+            return snapshotInfo;
         }));
     }
 

server/src/main/java/org/elasticsearch/search/SearchService.java

@@ -134,7 +134,7 @@ public class SearchService extends AbstractLifecycleComponent implements IndexEv
 
     /**
      * Enables low-level, frequent search cancellation checks. Enabling low-level checks will make long running searches to react
-     * to the cancellation request faster. It will produce more cancellation checks but benchmarking has shown these did not 
+     * to the cancellation request faster. It will produce more cancellation checks but benchmarking has shown these did not
      * noticeably slow down searches.
      */
     public static final Setting<Boolean> LOW_LEVEL_CANCELLATION_SETTING =
@@ -341,7 +341,7 @@ public void executeQueryPhase(ShardSearchRequest request, SearchTask task, Actio
     }
 
     private <T> void runAsync(long id, Supplier<T> executable, ActionListener<T> listener) {
-        getExecutor(id).execute(ActionRunnable.wrap(listener, l -> l.onResponse(executable.get())));
+        getExecutor(id).execute(ActionRunnable.supply(listener, executable::get));
     }
 
     private SearchPhaseResult executeQueryPhase(ShardSearchRequest request, SearchTask task) throws Exception {
@@ -1053,7 +1053,7 @@ private void rewriteShardRequest(ShardSearchRequest request, ActionListener<Shar
         Executor executor = getExecutor(shard);
         ActionListener<Rewriteable> actionListener = ActionListener.wrap(r ->
                 // now we need to check if there is a pending refresh and register
-                shard.awaitShardSearchActive(b -> executor.execute(ActionRunnable.wrap(listener, l -> l.onResponse(request)))),
+                shard.awaitShardSearchActive(b -> executor.execute(ActionRunnable.supply(listener, () -> request))),
             listener::onFailure);
         // we also do rewrite on the coordinating node (TransportSearchService) but we also need to do it here for BWC as well as
         // AliasFilters that might need to be rewritten. These are edge-cases but we are every efficient doing the rewrite here so it's not

test/framework/src/main/java/org/elasticsearch/repositories/AbstractThirdPartyRepositoryTestCase.java

@@ -82,13 +82,7 @@ public void tearDown() throws Exception {
     private void deleteAndAssertEmpty(BlobPath path) throws Exception {
         final BlobStoreRepository repo = getRepository();
         final PlainActionFuture<Void> future = PlainActionFuture.newFuture();
-        repo.threadPool().generic().execute(new ActionRunnable<>(future) {
-            @Override
-            protected void doRun() throws Exception {
-                repo.blobStore().blobContainer(path).delete();
-                future.onResponse(null);
-            }
-        });
+        repo.threadPool().generic().execute(ActionRunnable.run(future, () -> repo.blobStore().blobContainer(path).delete()));
         future.actionGet();
         final BlobPath parent = path.parent();
         if (parent == null) {
@@ -147,19 +141,15 @@ public void testListChildren() throws Exception {
         final PlainActionFuture<Void> future = PlainActionFuture.newFuture();
         final Executor genericExec = repo.threadPool().generic();
         final int testBlobLen = randomIntBetween(1, 100);
-        genericExec.execute(new ActionRunnable<>(future) {
-            @Override
-            protected void doRun() throws Exception {
-                final BlobStore blobStore = repo.blobStore();
-                blobStore.blobContainer(repo.basePath().add("foo"))
-                    .writeBlob("nested-blob", new ByteArrayInputStream(randomByteArrayOfLength(testBlobLen)), testBlobLen, false);
-                blobStore.blobContainer(repo.basePath().add("foo").add("nested"))
-                    .writeBlob("bar", new ByteArrayInputStream(randomByteArrayOfLength(testBlobLen)), testBlobLen, false);
-                blobStore.blobContainer(repo.basePath().add("foo").add("nested2"))
-                    .writeBlob("blub", new ByteArrayInputStream(randomByteArrayOfLength(testBlobLen)), testBlobLen, false);
-                future.onResponse(null);
-            }
-        });
+        genericExec.execute(ActionRunnable.run(future, () -> {
+            final BlobStore blobStore = repo.blobStore();
+            blobStore.blobContainer(repo.basePath().add("foo"))
+                .writeBlob("nested-blob", new ByteArrayInputStream(randomByteArrayOfLength(testBlobLen)), testBlobLen, false);
+            blobStore.blobContainer(repo.basePath().add("foo").add("nested"))
+                .writeBlob("bar", new ByteArrayInputStream(randomByteArrayOfLength(testBlobLen)), testBlobLen, false);
+            blobStore.blobContainer(repo.basePath().add("foo").add("nested2"))
+                .writeBlob("blub", new ByteArrayInputStream(randomByteArrayOfLength(testBlobLen)), testBlobLen, false);
+        }));
         future.actionGet();
         assertChildren(repo.basePath(), Collections.singleton("foo"));
         assertBlobsByPrefix(repo.basePath(), "fo", Collections.emptyMap());
@@ -245,37 +235,27 @@ protected void assertCleanupResponse(CleanupRepositoryResponse response, long by
 
     private void createDanglingIndex(final BlobStoreRepository repo, final Executor genericExec) throws Exception {
         final PlainActionFuture<Void> future = PlainActionFuture.newFuture();
-        genericExec.execute(new ActionRunnable<>(future) {
-            @Override
-            protected void doRun() throws Exception {
-                final BlobStore blobStore = repo.blobStore();
-                blobStore.blobContainer(repo.basePath().add("indices").add("foo"))
-                    .writeBlob("bar", new ByteArrayInputStream(new byte[3]), 3, false);
-                for (String prefix : Arrays.asList("snap-", "meta-")) {
-                    blobStore.blobContainer(repo.basePath())
-                        .writeBlob(prefix + "foo.dat", new ByteArrayInputStream(new byte[3]), 3, false);
-                }
-                future.onResponse(null);
+        genericExec.execute(ActionRunnable.run(future, () -> {
+            final BlobStore blobStore = repo.blobStore();
+            blobStore.blobContainer(repo.basePath().add("indices").add("foo"))
+                .writeBlob("bar", new ByteArrayInputStream(new byte[3]), 3, false);
+            for (String prefix : Arrays.asList("snap-", "meta-")) {
+                blobStore.blobContainer(repo.basePath()).writeBlob(prefix + "foo.dat", new ByteArrayInputStream(new byte[3]), 3, false);
             }
-        });
+        }));
         future.actionGet();
         assertTrue(assertCorruptionVisible(repo, genericExec));
     }
 
     protected boolean assertCorruptionVisible(BlobStoreRepository repo, Executor executor) throws Exception {
         final PlainActionFuture<Boolean> future = PlainActionFuture.newFuture();
-        executor.execute(new ActionRunnable<>(future) {
-            @Override
-            protected void doRun() throws Exception {
-                final BlobStore blobStore = repo.blobStore();
-                future.onResponse(
-                    blobStore.blobContainer(repo.basePath().add("indices")).children().containsKey("foo")
-                        && BlobStoreTestUtil.blobExists(blobStore.blobContainer(repo.basePath().add("indices").add("foo")), "bar")
-                        && BlobStoreTestUtil.blobExists(blobStore.blobContainer(repo.basePath()), "meta-foo.dat")
-                        && BlobStoreTestUtil.blobExists(blobStore.blobContainer(repo.basePath()), "snap-foo.dat")
-                );
-            }
-        });
+        executor.execute(ActionRunnable.supply(future, () -> {
+            final BlobStore blobStore = repo.blobStore();
+            return blobStore.blobContainer(repo.basePath().add("indices")).children().containsKey("foo")
+                && BlobStoreTestUtil.blobExists(blobStore.blobContainer(repo.basePath().add("indices").add("foo")), "bar")
+                && BlobStoreTestUtil.blobExists(blobStore.blobContainer(repo.basePath()), "meta-foo.dat")
+                && BlobStoreTestUtil.blobExists(blobStore.blobContainer(repo.basePath()), "snap-foo.dat");
+        }));
         return future.actionGet();
     }
 
@@ -300,13 +280,8 @@ protected void assertChildren(BlobPath path, Collection<String> children) throws
     private Set<String> listChildren(BlobPath path) {
         final PlainActionFuture<Set<String>> future = PlainActionFuture.newFuture();
         final BlobStoreRepository repository = getRepository();
-        repository.threadPool().generic().execute(new ActionRunnable<>(future) {
-            @Override
-            protected void doRun() throws Exception {
-                final BlobStore blobStore = repository.blobStore();
-                future.onResponse(blobStore.blobContainer(path).children().keySet());
-            }
-        });
+        repository.threadPool().generic().execute(
+            ActionRunnable.supply(future, () -> repository.blobStore().blobContainer(path).children().keySet()));
         return future.actionGet();
     }