remove thread pool

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java

@@ -29,8 +29,6 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeoutException;
 import java.util.function.Function;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -1518,12 +1516,12 @@ public interface AsyncAdmin {
    * @param serverNames the given list of region servers
    * @param <S>         the type of the asynchronous stub
    * @param <R>         the type of the return value
-   * @return a list of return values of the protobuf rpc call, wrapped by a {@link CompletableFuture}.
+   * @return Map of each region server to its result of the protobuf rpc call, wrapped by a
+   *         {@link CompletableFuture}.
    * @see ServiceCaller
    */
-  <S, R> CompletableFuture<List<R>> coprocessorService(Function<RpcChannel, S> stubMaker,
+  <S, R> CompletableFuture<Map<ServerName, Object>> coprocessorService(
-    ServiceCaller<S, R> callable, List<ServerName> serverNames) throws ExecutionException,
+    Function<RpcChannel, S> stubMaker, ServiceCaller<S, R> callable, List<ServerName> serverNames);
-    InterruptedException, TimeoutException;
 
   /**
    * List all the dead region servers.

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java

@@ -23,9 +23,7 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeoutException;
 import java.util.function.Function;
 import java.util.regex.Pattern;
 import org.apache.hadoop.hbase.CacheEvictionStats;
@@ -809,9 +807,8 @@ class AsyncHBaseAdmin implements AsyncAdmin {
   }
 
   @Override
-  public <S, R> CompletableFuture<List<R>> coprocessorService(Function<RpcChannel, S> stubMaker,
+  public <S, R> CompletableFuture<Map<ServerName, Object>> coprocessorService(
-    ServiceCaller<S, R> callable, List<ServerName> serverNames)
+    Function<RpcChannel, S> stubMaker, ServiceCaller<S, R> callable, List<ServerName> serverNames) {
-    throws ExecutionException, InterruptedException, TimeoutException {
     return wrap(rawAdmin.coprocessorService(stubMaker, callable, serverNames));
   }
 

hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncHBaseAdmin.java

@@ -36,11 +36,7 @@ import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
@@ -99,8 +95,8 @@ import org.apache.hadoop.hbase.snapshot.SnapshotCreationException;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.ForeignExceptionUtil;
+import org.apache.hadoop.hbase.util.FutureUtils;
 import org.apache.hadoop.hbase.util.Pair;
-import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -3511,41 +3507,28 @@ class RawAsyncHBaseAdmin implements AsyncAdmin {
   }
 
   @Override
-  public <S, R> CompletableFuture<List<R>> coprocessorService(Function<RpcChannel, S> stubMaker,
+  public <S, R> CompletableFuture<Map<ServerName, Object>> coprocessorService(
-    ServiceCaller<S, R> callable, List<ServerName> serverNames)
+    Function<RpcChannel, S> stubMaker, ServiceCaller<S, R> callable, List<ServerName> serverNames) {
-    throws ExecutionException, InterruptedException, TimeoutException {
+    CompletableFuture<Map<ServerName, Object>> future = new CompletableFuture<>();
-    List<S> stubs = new ArrayList<>(serverNames.size());
+    Map<ServerName, Object> resultMap = new HashMap<>();
-    for(ServerName serverName : serverNames) {
+    for (ServerName rs : serverNames) {
-      RegionServerCoprocessorRpcChannelImpl channel = new RegionServerCoprocessorRpcChannelImpl(
+      FutureUtils.addListener(coprocessorService(stubMaker, callable, rs), (r, e) -> {
-        this.<Message> newServerCaller().serverName(serverName));
+        boolean done;
-      S stub = stubMaker.apply(channel);
+        synchronized (resultMap) {
-      stubs.add(stub);
+          if (e != null) {
-    }
+            resultMap.put(rs, e);
-
-    return CompletableFuture.supplyAsync(() -> {
-      ExecutorService executorService = Executors.newFixedThreadPool(serverNames.size(),
-        new ThreadFactoryBuilder().setNameFormat("coproc-service-%d").setDaemon(true).build());
-      List<CompletableFuture<R>> completableFutureList = new ArrayList<>();
-      for (S stub : stubs) {
-        CompletableFuture<R> future = new CompletableFuture<>();
-        completableFutureList.add(future);
-        ClientCoprocessorRpcController controller = new ClientCoprocessorRpcController();
-        executorService.execute(() -> callable.call(stub, controller, resp -> {
-          if (controller.failed()) {
-            future.completeExceptionally(controller.getFailed());
           } else {
-            future.complete(resp);
+            resultMap.put(rs, r);
           }
-        }));
+          done = resultMap.size() == serverNames.size();
         }
-      List<R> listValues = new ArrayList<>(serverNames.size());
+        if (done) {
-      for(CompletableFuture<R> completableFuture : completableFutureList) {
+          future.complete(resultMap);
-        listValues.add(completableFuture.join());
         }
-      executorService.shutdownNow();
-      return listValues;
       });
     }
+    return future;
+  }
 
   @Override
   public CompletableFuture<List<ServerName>> clearDeadServers(List<ServerName> servers) {

hbase-endpoint/src/test/java/org/apache/hadoop/hbase/coprocessor/TestAsyncRegionServersCoprocessorEndpoint.java

@@ -19,12 +19,13 @@ package org.apache.hadoop.hbase.coprocessor;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
 
 import java.io.FileNotFoundException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
 import org.apache.hadoop.hbase.HBaseClassTestRule;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.ServerName;
@@ -46,7 +47,6 @@ import org.apache.hbase.thirdparty.com.google.protobuf.RpcCallback;
 import org.apache.hbase.thirdparty.com.google.protobuf.RpcController;
 import org.apache.hbase.thirdparty.com.google.protobuf.Service;
 
-import org.apache.hadoop.hbase.shaded.coprocessor.protobuf.generated.DummyRegionServerEndpointProtos;
 import org.apache.hadoop.hbase.shaded.coprocessor.protobuf.generated.DummyRegionServerEndpointProtos.DummyRequest;
 import org.apache.hadoop.hbase.shaded.coprocessor.protobuf.generated.DummyRegionServerEndpointProtos.DummyResponse;
 import org.apache.hadoop.hbase.shaded.coprocessor.protobuf.generated.DummyRegionServerEndpointProtos.DummyService;
@@ -76,54 +76,47 @@ public class TestAsyncRegionServersCoprocessorEndpoint extends TestAsyncAdminBas
   }
 
   @Test
-  public void testRegionServerCoprocessorServiceWithMultipleServers() throws Exception {
+  public void testRegionServersCoprocessorService()
-    final List<JVMClusterUtil.RegionServerThread> regionServerThreads = TEST_UTIL.getHBaseCluster().getRegionServerThreads();
+    throws ExecutionException, InterruptedException {
+    final List<JVMClusterUtil.RegionServerThread> regionServerThreads =
+      TEST_UTIL.getHBaseCluster().getRegionServerThreads();
     List<ServerName> serverNames = new ArrayList<>();
-    for (JVMClusterUtil.RegionServerThread t : regionServerThreads) {
+    regionServerThreads.forEach(t -> serverNames.add(t.getRegionServer().getServerName()));
-      serverNames.add(t.getRegionServer().getServerName());
-    }
-    DummyRegionServerEndpointProtos.DummyRequest request =
-      DummyRegionServerEndpointProtos.DummyRequest.getDefaultInstance();
-    List<DummyResponse> responses =
-      admin.<DummyRegionServerEndpointProtos.DummyService.Stub,
-          DummyRegionServerEndpointProtos.DummyResponse> coprocessorService(
-          DummyRegionServerEndpointProtos.DummyService::newStub,
-          (s, c, done) -> s.dummyCall(c, request, done), serverNames)
-        .get();
 
-    assertEquals(responses.size(), serverNames.size());
+    DummyRequest request = DummyRequest.getDefaultInstance();
-    for (DummyResponse response : responses) {
+    Map<ServerName, Object> resultMap =
-      assertEquals(DUMMY_VALUE, response.getValue());
+      admin.<DummyService.Stub, DummyResponse> coprocessorService(DummyService::newStub,
-    }
+        (s, c, done) -> s.dummyCall(c, request, done), serverNames).get();
+
+    resultMap.forEach((k, v) -> {
+      assertTrue(v instanceof DummyResponse);
+      DummyResponse resp = (DummyResponse) v;
+      assertEquals(DUMMY_VALUE, resp.getValue());
+    });
   }
 
   @Test
-  public void testRegionServerCoprocessorServiceErrorWithMultipleServers() throws Exception {
+  public void testRegionServerCoprocessorsServiceError()
-    final List<JVMClusterUtil.RegionServerThread> regionServerThreads = TEST_UTIL.getHBaseCluster().getRegionServerThreads();
+    throws ExecutionException, InterruptedException {
+    final List<JVMClusterUtil.RegionServerThread> regionServerThreads =
+      TEST_UTIL.getHBaseCluster().getRegionServerThreads();
     List<ServerName> serverNames = new ArrayList<>();
-    for (JVMClusterUtil.RegionServerThread t : regionServerThreads) {
+    regionServerThreads.forEach(t -> serverNames.add(t.getRegionServer().getServerName()));
-      serverNames.add(t.getRegionServer().getServerName());
+
-    }
+    DummyRequest request = DummyRequest.getDefaultInstance();
-    DummyRegionServerEndpointProtos.DummyRequest request =
+    Map<ServerName, Object> resultMap =
-      DummyRegionServerEndpointProtos.DummyRequest.getDefaultInstance();
+      admin.<DummyService.Stub, DummyResponse> coprocessorService(DummyService::newStub,
-    try {
+        (s, c, done) -> s.dummyThrow(c, request, done), serverNames).get();
-      admin.<DummyRegionServerEndpointProtos.DummyService.Stub,
+
-          DummyRegionServerEndpointProtos.DummyResponse> coprocessorService(
+    resultMap.forEach((k, v) -> {
-          DummyRegionServerEndpointProtos.DummyService::newStub,
+      assertTrue(v instanceof RetriesExhaustedException);
-          (s, c, done) -> s.dummyThrow(c, request, done), serverNames)
+      Throwable e = (Throwable) v;
-        .get();
+      assertTrue(e.getMessage().contains(WHAT_TO_THROW.getClass().getName().trim()));
-      fail("Should have thrown an exception");
+    });
-    } catch (Exception e) {
-      assertTrue(e.getCause() instanceof RetriesExhaustedException);
-      assertTrue(e.getCause().getMessage().contains(WHAT_TO_THROW.getClass().getName().trim()));
-    }
   }
 
   public static class DummyRegionServerEndpoint extends DummyService
     implements RegionServerCoprocessor {
-    public DummyRegionServerEndpoint() {
-
-    }
     @Override
     public Iterable<Service> getServices() {
       return Collections.singleton(this);