Merge branch 'main' into fuse_mv_min_max

Author: nik9000

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/MultiClusterSpecIT.java

@@ -15,7 +15,6 @@
 import org.elasticsearch.Version;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
-import org.elasticsearch.client.ResponseListener;
 import org.elasticsearch.client.RestClient;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.IOUtils;
@@ -39,8 +38,6 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
@@ -254,7 +251,10 @@ static RestClient twoClients(RestClient localClient, RestClient remoteClient) th
                     return bulkClient.performRequest(request);
                 } else {
                     Request[] clones = cloneRequests(request, 2);
-                    return runInParallel(localClient, remoteClient, clones);
+                    Response resp1 = remoteClient.performRequest(clones[0]);
+                    Response resp2 = localClient.performRequest(clones[1]);
+                    assertEquals(resp1.getStatusLine().getStatusCode(), resp2.getStatusLine().getStatusCode());
+                    return resp2;
                 }
         });
         doAnswer(invocation -> {
@@ -289,44 +289,6 @@ static Request[] cloneRequests(Request orig, int numClones) throws IOException {
         return clones;
     }
 
-    /**
-     * Run {@link #cloneRequests cloned} requests in parallel.
-     */
-    static Response runInParallel(RestClient localClient, RestClient remoteClient, Request[] clones) throws Throwable {
-        CompletableFuture<Response> remoteResponse = new CompletableFuture<>();
-        CompletableFuture<Response> localResponse = new CompletableFuture<>();
-        remoteClient.performRequestAsync(clones[0], new ResponseListener() {
-            @Override
-            public void onSuccess(Response response) {
-                remoteResponse.complete(response);
-            }
-
-            @Override
-            public void onFailure(Exception exception) {
-                remoteResponse.completeExceptionally(exception);
-            }
-        });
-        localClient.performRequestAsync(clones[1], new ResponseListener() {
-            @Override
-            public void onSuccess(Response response) {
-                localResponse.complete(response);
-            }
-
-            @Override
-            public void onFailure(Exception exception) {
-                localResponse.completeExceptionally(exception);
-            }
-        });
-        try {
-            Response remote = remoteResponse.get();
-            Response local = localResponse.get();
-            assertEquals(remote.getStatusLine().getStatusCode(), local.getStatusLine().getStatusCode());
-            return local;
-        } catch (ExecutionException e) {
-            throw e.getCause();
-        }
-    }
-
     /**
      * Convert FROM employees ... => FROM *:employees,employees
      */