Remove Custom Listeners from SnapshotsService (#37629)

* Remove Custom Listeners from SnapshotsService

Motivations:
    * Shorten the code some more
    * Use ActionListener#wrap to get easy to reason about behavior in failure scenarios
    * Remove duplication in the logic of handling snapshot completion listeners (listeners removing themselves and comparing snapshots to their targets)
        * Also here, move all listener handling into `SnapshotsService` and remove custom listener class by putting listeners in a map

Author: original-brownbear

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/create/TransportCreateSnapshotAction.java

@@ -28,8 +28,6 @@
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.snapshots.Snapshot;
-import org.elasticsearch.snapshots.SnapshotInfo;
 import org.elasticsearch.snapshots.SnapshotsService;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
@@ -72,37 +70,13 @@ protected ClusterBlockException checkBlock(CreateSnapshotRequest request, Cluste
 
     @Override
     protected void masterOperation(final CreateSnapshotRequest request, ClusterState state,
-                                   final ActionListener<CreateSnapshotResponse> listener) {
-        snapshotsService.createSnapshot(request, new SnapshotsService.CreateSnapshotListener() {
-            @Override
-            public void onResponse(Snapshot snapshotCreated) {
-                if (request.waitForCompletion()) {
-                    snapshotsService.addListener(new SnapshotsService.SnapshotCompletionListener() {
-                        @Override
-                        public void onSnapshotCompletion(Snapshot snapshot, SnapshotInfo snapshotInfo) {
-                            if (snapshotCreated.equals(snapshot)) {
-                                listener.onResponse(new CreateSnapshotResponse(snapshotInfo));
-                                snapshotsService.removeListener(this);
-                            }
-                        }
-
-                        @Override
-                        public void onSnapshotFailure(Snapshot snapshot, Exception e) {
-                            if (snapshotCreated.equals(snapshot)) {
-                                listener.onFailure(e);
-                                snapshotsService.removeListener(this);
-                            }
-                        }
-                    });
-                } else {
-                    listener.onResponse(new CreateSnapshotResponse());
-                }
-            }
-
-            @Override
-            public void onFailure(Exception e) {
-                listener.onFailure(e);
-            }
-        });
+        final ActionListener<CreateSnapshotResponse> listener) {
+        if (request.waitForCompletion()) {
+            snapshotsService.executeSnapshot(request,
+                ActionListener.wrap(snapshotInfo-> listener.onResponse(new CreateSnapshotResponse(snapshotInfo)), listener::onFailure));
+        } else {
+            snapshotsService.createSnapshot(request,
+                ActionListener.wrap(snapshot -> listener.onResponse(new CreateSnapshotResponse()), listener::onFailure));
+        }
     }
 }

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/delete/TransportDeleteSnapshotAction.java

@@ -67,16 +67,7 @@ protected ClusterBlockException checkBlock(DeleteSnapshotRequest request, Cluste
     @Override
     protected void masterOperation(final DeleteSnapshotRequest request, ClusterState state,
                                    final ActionListener<AcknowledgedResponse> listener) {
-        snapshotsService.deleteSnapshot(request.repository(), request.snapshot(), new SnapshotsService.DeleteSnapshotListener() {
-            @Override
-            public void onResponse() {
-                listener.onResponse(new AcknowledgedResponse(true));
-            }
-
-            @Override
-            public void onFailure(Exception e) {
-                listener.onFailure(e);
-            }
-        }, false);
+        snapshotsService.deleteSnapshot(request.repository(), request.snapshot(),
+            ActionListener.wrap(v -> listener.onResponse(new AcknowledgedResponse(true)), listener::onFailure), false);
     }
 }