elastic · original-brownbear · Jun 10, 2021 · Jun 9, 2021
diff --git a/...st-http/src/test/java/org/elasticsearch/http/BlockedSearcherRestCancellationTestCase.java b/...st-http/src/test/java/org/elasticsearch/http/BlockedSearcherRestCancellationTestCase.java
@@ -12,7 +12,6 @@
 import org.elasticsearch.client.Cancellable;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
-import org.elasticsearch.client.ResponseListener;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.common.settings.Setting;
@@ -41,6 +40,7 @@
 import java.util.function.Function;
 
 import static java.util.Collections.singletonList;
+import static org.elasticsearch.action.support.ActionTestUtils.wrapAsRestResponseListener;
 import static org.elasticsearch.test.TaskAssertions.assertAllCancellableTasksAreCancelled;
 import static org.elasticsearch.test.TaskAssertions.assertAllTasksHaveFinished;
 import static org.elasticsearch.test.TaskAssertions.awaitTaskWithPrefix;
@@ -92,19 +92,9 @@ void runTest(Request request, String actionPrefix) throws Exception {
                 releasables.add(searcherBlock::release);
             }
 
-            final PlainActionFuture<Void> future = new PlainActionFuture<>();
+            final PlainActionFuture<Response> future = new PlainActionFuture<>();
             logger.info("--> sending request");
-            final Cancellable cancellable = getRestClient().performRequestAsync(request, new ResponseListener() {
-                @Override
-                public void onSuccess(Response response) {
-                    future.onResponse(null);
-                }
-
-                @Override
-                public void onFailure(Exception exception) {
-                    future.onFailure(exception);
-                }
-            });
+            final Cancellable cancellable = getRestClient().performRequestAsync(request, wrapAsRestResponseListener(future));
 
             awaitTaskWithPrefix(actionPrefix);
 

diff --git a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/ClusterStateRestCancellationIT.java b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/ClusterStateRestCancellationIT.java
@@ -15,7 +15,6 @@
 import org.elasticsearch.client.Cancellable;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
-import org.elasticsearch.client.ResponseListener;
 import org.elasticsearch.cluster.AbstractDiffable;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
@@ -33,6 +32,7 @@
 import java.util.concurrent.CancellationException;
 import java.util.function.UnaryOperator;
 
+import static org.elasticsearch.action.support.ActionTestUtils.wrapAsRestResponseListener;
 import static org.elasticsearch.test.TaskAssertions.awaitTaskWithPrefix;
 
 public class ClusterStateRestCancellationIT extends HttpSmokeTestCase {
@@ -75,19 +75,9 @@ public void testClusterStateRestCancellation() throws Exception {
             clusterStateRequest.addParameter("local", "true");
         }
 
-        final PlainActionFuture<Void> future = new PlainActionFuture<>();
+        final PlainActionFuture<Response> future = new PlainActionFuture<>();
         logger.info("--> sending cluster state request");
-        final Cancellable cancellable = getRestClient().performRequestAsync(clusterStateRequest, new ResponseListener() {
-            @Override
-            public void onSuccess(Response response) {
-                future.onResponse(null);
-            }
-
-            @Override
-            public void onFailure(Exception exception) {
-                future.onFailure(exception);
-            }
-        });
+        final Cancellable cancellable = getRestClient().performRequestAsync(clusterStateRequest, wrapAsRestResponseListener(future));
 
         awaitTaskWithPrefix(ClusterStateAction.NAME);
 

diff --git a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/ClusterStatsRestCancellationIT.java b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/ClusterStatsRestCancellationIT.java
@@ -14,7 +14,6 @@
 import org.elasticsearch.client.Cancellable;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
-import org.elasticsearch.client.ResponseListener;
 import org.elasticsearch.cluster.routing.allocation.DiskThresholdSettings;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
@@ -44,6 +43,7 @@
 import java.util.function.Function;
 
 import static java.util.Collections.singletonList;
+import static org.elasticsearch.action.support.ActionTestUtils.wrapAsRestResponseListener;
 import static org.elasticsearch.test.TaskAssertions.assertAllCancellableTasksAreCancelled;
 import static org.elasticsearch.test.TaskAssertions.assertAllTasksHaveFinished;
 import static org.elasticsearch.test.TaskAssertions.awaitTaskWithPrefix;
@@ -100,19 +100,9 @@ public void testClusterStateRestCancellation() throws Exception {
 
             final Request clusterStatsRequest = new Request(HttpGet.METHOD_NAME, "/_cluster/stats");
 
-            final PlainActionFuture<Void> future = new PlainActionFuture<>();
+            final PlainActionFuture<Response> future = new PlainActionFuture<>();
             logger.info("--> sending cluster state request");
-            final Cancellable cancellable = getRestClient().performRequestAsync(clusterStatsRequest, new ResponseListener() {
-                @Override
-                public void onSuccess(Response response) {
-                    future.onResponse(null);
-                }
-
-                @Override
-                public void onFailure(Exception exception) {
-                    future.onFailure(exception);
-                }
-            });
+            final Cancellable cancellable = getRestClient().performRequestAsync(clusterStatsRequest, wrapAsRestResponseListener(future));
 
             awaitTaskWithPrefix(ClusterStatsAction.NAME);
 

diff --git a/...oke-test-http/src/test/java/org/elasticsearch/http/IndicesRecoveryRestCancellationIT.java b/...oke-test-http/src/test/java/org/elasticsearch/http/IndicesRecoveryRestCancellationIT.java
@@ -16,7 +16,6 @@
 import org.elasticsearch.client.Cancellable;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
-import org.elasticsearch.client.ResponseListener;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
 
@@ -25,6 +24,7 @@
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.Semaphore;
 
+import static org.elasticsearch.action.support.ActionTestUtils.wrapAsRestResponseListener;
 import static org.elasticsearch.test.TaskAssertions.assertAllCancellableTasksAreCancelled;
 import static org.elasticsearch.test.TaskAssertions.assertAllTasksHaveFinished;
 import static org.elasticsearch.test.TaskAssertions.awaitTaskWithPrefix;
@@ -68,19 +68,9 @@ private void runTest(Request request) throws Exception {
                 releasables.add(operationBlock::release);
             }
 
-            final PlainActionFuture<Void> future = new PlainActionFuture<>();
+            final PlainActionFuture<Response> future = new PlainActionFuture<>();
             logger.info("--> sending request");
-            final Cancellable cancellable = getRestClient().performRequestAsync(request, new ResponseListener() {
-                @Override
-                public void onSuccess(Response response) {
-                    future.onResponse(null);
-                }
-
-                @Override
-                public void onFailure(Exception exception) {
-                    future.onFailure(exception);
-                }
-            });
+            final Cancellable cancellable = getRestClient().performRequestAsync(request, wrapAsRestResponseListener(future));
 
             awaitTaskWithPrefix(RecoveryAction.NAME);
 

diff --git a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/RestGetMappingsCancellationIT.java b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/RestGetMappingsCancellationIT.java
@@ -15,7 +15,6 @@
 import org.elasticsearch.client.Cancellable;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
-import org.elasticsearch.client.ResponseListener;
 import org.elasticsearch.cluster.AckedClusterStateUpdateTask;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ack.AckedRequest;
@@ -31,6 +30,7 @@
 import java.util.concurrent.CancellationException;
 import java.util.function.Function;
 
+import static org.elasticsearch.action.support.ActionTestUtils.wrapAsRestResponseListener;
 import static org.elasticsearch.test.TaskAssertions.assertAllCancellableTasksAreCancelled;
 import static org.elasticsearch.test.TaskAssertions.assertAllTasksHaveFinished;
 import static org.elasticsearch.test.TaskAssertions.awaitTaskWithPrefix;
@@ -64,18 +64,8 @@ public void testGetMappingsCancellation() throws Exception {
         });
 
         final Request request = new Request(HttpGet.METHOD_NAME, "/test/_mappings");
-        final PlainActionFuture<Void> future = new PlainActionFuture<>();
-        final Cancellable cancellable = getRestClient().performRequestAsync(request, new ResponseListener() {
-            @Override
-            public void onSuccess(Response response) {
-                future.onResponse(null);
-            }
-
-            @Override
-            public void onFailure(Exception exception) {
-                future.onFailure(exception);
-            }
-        });
+        final PlainActionFuture<Response> future = new PlainActionFuture<>();
+        final Cancellable cancellable = getRestClient().performRequestAsync(request, wrapAsRestResponseListener(future));
 
         assertThat(future.isDone(), equalTo(false));
         awaitTaskWithPrefix(actionName);

diff --git a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/SearchRestCancellationIT.java b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/SearchRestCancellationIT.java
@@ -19,11 +19,11 @@
 import org.elasticsearch.action.search.MultiSearchRequest;
 import org.elasticsearch.action.search.SearchAction;
 import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.client.Cancellable;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
-import org.elasticsearch.client.ResponseListener;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.common.xcontent.XContentType;
@@ -49,13 +49,12 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CancellationException;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 
+import static org.elasticsearch.action.support.ActionTestUtils.wrapAsRestResponseListener;
 import static org.elasticsearch.index.query.QueryBuilders.scriptQuery;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
 import static org.hamcrest.Matchers.greaterThan;
@@ -93,28 +92,15 @@ void verifyCancellationDuringQueryPhase(String searchAction, Request searchReque
         List<ScriptedBlockPlugin> plugins = initBlockFactory();
         indexTestData();
 
-        CountDownLatch latch = new CountDownLatch(1);
-        AtomicReference<Exception> error = new AtomicReference<>();
-        Cancellable cancellable = getRestClient().performRequestAsync(searchRequest, new ResponseListener() {
-            @Override
-            public void onSuccess(Response response) {
-                latch.countDown();
-            }
-
-            @Override
-            public void onFailure(Exception exception) {
-                error.set(exception);
-                latch.countDown();
-            }
-        });
+        PlainActionFuture<Response> future = PlainActionFuture.newFuture();
+        Cancellable cancellable = getRestClient().performRequestAsync(searchRequest, wrapAsRestResponseListener(future));
 
         awaitForBlock(plugins);
         cancellable.cancel();
         ensureSearchTaskIsCancelled(searchAction, nodeIdToName::get);
 
         disableBlocks(plugins);
-        latch.await();
-        assertThat(error.get(), instanceOf(CancellationException.class));
+        expectThrows(CancellationException.class, future::actionGet);
     }
 
     public void testAutomaticCancellationDuringFetchPhase() throws Exception {
@@ -142,28 +128,15 @@ void verifyCancellationDuringFetchPhase(String searchAction, Request searchReque
         List<ScriptedBlockPlugin> plugins = initBlockFactory();
         indexTestData();
 
-        CountDownLatch latch = new CountDownLatch(1);
-        AtomicReference<Exception> error = new AtomicReference<>();
-        Cancellable cancellable = getRestClient().performRequestAsync(searchRequest, new ResponseListener() {
-            @Override
-            public void onSuccess(Response response) {
-                latch.countDown();
-            }
-
-            @Override
-            public void onFailure(Exception exception) {
-                error.set(exception);
-                latch.countDown();
-            }
-        });
+        PlainActionFuture<Response> future = PlainActionFuture.newFuture();
+        Cancellable cancellable = getRestClient().performRequestAsync(searchRequest, wrapAsRestResponseListener(future));
 
         awaitForBlock(plugins);
         cancellable.cancel();
         ensureSearchTaskIsCancelled(searchAction, nodeIdToName::get);
 
         disableBlocks(plugins);
-        latch.await();
-        assertThat(error.get(), instanceOf(CancellationException.class));
+        expectThrows(CancellationException.class, future::actionGet);
     }
 
     private static Map<String, String> readNodesInfo() {

diff --git a/...est-http/src/test/java/org/elasticsearch/http/snapshots/AbstractSnapshotRestTestCase.java b/...est-http/src/test/java/org/elasticsearch/http/snapshots/AbstractSnapshotRestTestCase.java
@@ -0,0 +1,26 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.http.snapshots;
+
+import org.elasticsearch.common.util.CollectionUtils;
+import org.elasticsearch.http.HttpSmokeTestCase;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.snapshots.mockstore.MockRepository;
+import org.elasticsearch.test.ESIntegTestCase;
+
+import java.util.Collection;
+
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0, numClientNodes = 0)
+public abstract class AbstractSnapshotRestTestCase extends HttpSmokeTestCase {
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return CollectionUtils.appendToCopy(super.nodePlugins(), MockRepository.Plugin.class);
+    }
+}
diff --git a/.../http/RestGetSnapshotsCancellationIT.java → ...shots/RestGetSnapshotsCancellationIT.java b/.../http/RestGetSnapshotsCancellationIT.java → ...shots/RestGetSnapshotsCancellationIT.java
@@ -6,76 +6,43 @@
  * Side Public License, v 1.
  */
 
-package org.elasticsearch.http;
+package org.elasticsearch.http.snapshots;
 
 import org.apache.http.client.methods.HttpGet;
 import org.elasticsearch.action.admin.cluster.snapshots.get.GetSnapshotsAction;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.client.Cancellable;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
-import org.elasticsearch.client.ResponseListener;
-import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.util.CollectionUtils;
-import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.snapshots.AbstractSnapshotIntegTestCase;
-import org.elasticsearch.snapshots.SnapshotState;
 import org.elasticsearch.snapshots.mockstore.MockRepository;
-import org.elasticsearch.test.ESIntegTestCase;
 
-import java.util.Collection;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.TimeUnit;
 
+import static org.elasticsearch.action.support.ActionTestUtils.wrapAsRestResponseListener;
 import static org.elasticsearch.test.TaskAssertions.assertAllCancellableTasksAreCancelled;
 import static org.elasticsearch.test.TaskAssertions.assertAllTasksHaveFinished;
 import static org.elasticsearch.test.TaskAssertions.awaitTaskWithPrefix;
-import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.hamcrest.core.IsEqual.equalTo;
 
-@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0, numClientNodes = 0)
-public class RestGetSnapshotsCancellationIT extends HttpSmokeTestCase {
-
-    @Override
-    protected Collection<Class<? extends Plugin>> nodePlugins() {
-        return CollectionUtils.appendToCopy(super.nodePlugins(), MockRepository.Plugin.class);
-    }
+public class RestGetSnapshotsCancellationIT extends AbstractSnapshotRestTestCase {
 
     public void testGetSnapshotsCancellation() throws Exception {
         internalCluster().startMasterOnlyNode();
         internalCluster().startDataOnlyNode();
         ensureStableCluster(2);
 
         final String repoName = "test-repo";
-        assertAcked(
-                client().admin().cluster().preparePutRepository(repoName)
-                        .setType("mock").setSettings(Settings.builder().put("location", randomRepoPath())));
-
-        final int snapshotCount = randomIntBetween(1, 5);
-        for (int i = 0; i < snapshotCount; i++) {
-            assertEquals(
-                    SnapshotState.SUCCESS,
-                    client().admin().cluster().prepareCreateSnapshot(repoName, "snapshot-" + i).setWaitForCompletion(true)
-                            .get().getSnapshotInfo().state()
-            );
-        }
+        AbstractSnapshotIntegTestCase.createRepository(logger, repoName, "mock");
+        AbstractSnapshotIntegTestCase.createNSnapshots(logger, repoName, randomIntBetween(1, 5));
 
         final MockRepository repository = AbstractSnapshotIntegTestCase.getRepositoryOnMaster(repoName);
         repository.setBlockOnAnyFiles();
 
         final Request request = new Request(HttpGet.METHOD_NAME, "/_snapshot/" + repoName + "/*");
-        final PlainActionFuture<Void> future = new PlainActionFuture<>();
-        final Cancellable cancellable = getRestClient().performRequestAsync(request, new ResponseListener() {
-            @Override
-            public void onSuccess(Response response) {
-                future.onResponse(null);
-            }
-
-            @Override
-            public void onFailure(Exception exception) {
-                future.onFailure(exception);
-            }
-        });
+        final PlainActionFuture<Response> future = new PlainActionFuture<>();
+        final Cancellable cancellable = getRestClient().performRequestAsync(request, wrapAsRestResponseListener(future));
 
         assertThat(future.isDone(), equalTo(false));
         awaitTaskWithPrefix(GetSnapshotsAction.NAME);