static Map<ServerLocation, Object> submitGetAll(
      Map<ServerLocation, Set> serverToFilterMap,
      List callableTasks, ClientMetadataService cms,
      LocalRegion region) {

    if (callableTasks != null && !callableTasks.isEmpty()) {
      Map<ServerLocation, Object> resultMap = new HashMap<>();
      List futures;
      try {
        futures = execService.invokeAll(callableTasks);
      } catch (InterruptedException e) {
        throw new InternalGemFireException(e.getMessage());
      }
      Iterator futureItr = futures.iterator();
      Iterator taskItr = callableTasks.iterator();
      while (futureItr.hasNext() && !execService.isShutdown() && !execService.isTerminated()) {
        Future fut = (Future) futureItr.next();
        SingleHopOperationCallable task = (SingleHopOperationCallable) taskItr.next();
        List keys = ((GetAllOpImpl) task.getOperation()).getKeyList();
        ServerLocation server = task.getServer();
        try {

          VersionedObjectList valuesFromServer = (VersionedObjectList) fut.get();
          valuesFromServer.setKeys(keys);

          for (VersionedObjectList.Iterator it = valuesFromServer.iterator(); it.hasNext();) {
            VersionedObjectList.Entry entry = it.next();
            Object key = entry.getKey();
            Object value = entry.getValue();
            if (!entry.isKeyNotOnServer()) {
              if (value instanceof Throwable) {
                logger.warn(String.format(
                    "%s: Caught the following exception attempting to get value for key=%s",
                    value, key),
                    (Throwable) value);
              }
            }
          }
          if (logger.isDebugEnabled()) {
            logger.debug("GetAllOp#got result from {}: {}", server, valuesFromServer);
          }
          resultMap.put(server, valuesFromServer);
        } catch (InterruptedException e) {
          throw new InternalGemFireException(e.getMessage());
        } catch (ExecutionException ee) {
          if (ee.getCause() instanceof ServerOperationException) {
            if (logger.isDebugEnabled()) {
              logger.debug("GetAllOp#ExecutionException.ServerOperationException: Caused by:",
                  ee.getCause());
            }
            throw (ServerOperationException) ee.getCause();
          } else if (ee.getCause() instanceof ServerConnectivityException) {
            if (logger.isDebugEnabled()) {
              logger.debug(
                  "GetAllOp#ExecutionException.ServerConnectivityException : Caused by :{} The failed server is: {}",
                  ee.getCause(), server);
            }
            try {
              cms = region.getCache().getClientMetadataService();
            } catch (CacheClosedException e) {
              return null;
            }
            cms.removeBucketServerLocation(server);
            cms.scheduleGetPRMetaData(region, false);
            resultMap.put(server, ee.getCause());
          } else {
            throw executionThrowable(ee.getCause());
          }
        }
      }
      return resultMap;
    }
    return null;
  }
execute bulk op (putAll or removeAll) on multiple PR servers, returning a map of the results.
Results are either a VersionedObjectList or a BulkOpPartialResultsException

@return the per-server results
geode-core/src/main/java/org/apache/geode/cache/client/internal/SingleHopClientExecutor.java