public static VersionedObjectList execute(ExecutablePool pool, Region region,
      Collection<Object> keys, EventID eventId, int retryAttempts, Object callbackArg) {
    final boolean isDebugEnabled = logger.isDebugEnabled();
    ClientMetadataService cms = ((InternalRegion) region).getCache().getClientMetadataService();

    Map<ServerLocation, Set> serverToFilterMap = cms.getServerToFilterMap(keys, region, true);

    if (serverToFilterMap == null || serverToFilterMap.isEmpty()) {
      AbstractOp op = new RemoveAllOpImpl(region, keys, eventId,
          ((PoolImpl) pool).getPRSingleHopEnabled(), callbackArg);
      op.initMessagePart();
      return (VersionedObjectList) pool.execute(op);
    }

    List callableTasks = constructAndGetRemoveAllTasks(region, eventId, serverToFilterMap,
        (PoolImpl) pool, callbackArg);

    if (isDebugEnabled) {
      logger.debug("RemoveAllOp#execute : Number of removeAll tasks is :{}", callableTasks.size());
    }
    HashMap<ServerLocation, RuntimeException> failedServers =
        new HashMap<>();
    PutAllPartialResult result = new PutAllPartialResult(keys.size());
    try {
      Map<ServerLocation, Object> results = SingleHopClientExecutor
          .submitBulkOp(callableTasks, cms,
              (LocalRegion) region, failedServers);
      for (Map.Entry<ServerLocation, Object> entry : results.entrySet()) {
        Object value = entry.getValue();
        if (value instanceof PutAllPartialResultException) {
          PutAllPartialResultException pap = (PutAllPartialResultException) value;
          if (isDebugEnabled) {
            logger.debug(
                "RemoveAll SingleHop encountered BulkOpPartialResultException exception: {}, failedServers are {}",
                pap, failedServers.keySet());
          }
          result.consolidate(pap.getResult());
        } else {
          if (value != null) {
            VersionedObjectList list = (VersionedObjectList) value;
            result.addKeysAndVersions(list);
          }
        }
      }
    } catch (RuntimeException ex) {
      logger.debug("single-hop removeAll encountered unexpected exception:", ex);
      throw ex;
    }

    if (!failedServers.isEmpty()) {
      if (retryAttempts == 0) {
        throw failedServers.values().iterator().next();
      }

      // if the partial result set doesn't already have keys (for tracking version tags)
      // then we need to gather up the keys that we know have succeeded so far and
      // add them to the partial result set
      if (result.getSucceededKeysAndVersions().size() == 0) {
        // if there're failed servers, we need to save the succeed keys in submitRemoveAll
        // if retry succeeded, everything is ok, otherwise, the saved "succeeded
        // keys" should be consolidated into PutAllPartialResultException
        // succeedKeySet is used to send back to client in PartialResult case
        // so it's not a must to use LinkedHashSet
        Set succeedKeySet = new LinkedHashSet();
        Set<ServerLocation> serverSet = serverToFilterMap.keySet();
        for (ServerLocation server : serverSet) {
          if (!failedServers.containsKey(server)) {
            succeedKeySet.addAll(serverToFilterMap.get(server));
          }
        }

        // save succeedKeys, but if retries all succeeded, discard the PutAllPartialResult
        result.addKeys(succeedKeySet);
      }

      // send maps for the failed servers one by one instead of merging
      // them into one big map. The reason is, we have to keep the same event
      // ids for each sub map. There is a unit test in PutAllCSDUnitTest for
      // the otherwise case.
      boolean oneSubMapRetryFailed = false;
      Set<ServerLocation> failedServerSet = failedServers.keySet();
      for (ServerLocation failedServer : failedServerSet) {
        RuntimeException savedRTE = failedServers.get(failedServer);
        if (savedRTE instanceof PutAllPartialResultException) {
          // will not retry for BulkOpPartialResultException
          // but it means at least one sub map ever failed
          oneSubMapRetryFailed = true;
          continue;
        }
        Collection<Object> newKeys = serverToFilterMap.get(failedServer);
        try {
          VersionedObjectList v =
              RemoveAllOp.execute(pool, region, newKeys, eventId, true, callbackArg);
          if (v == null) {
            result.addKeys(newKeys);
          } else {
            result.addKeysAndVersions(v);
          }
        } catch (PutAllPartialResultException pre) {
          oneSubMapRetryFailed = true;
          logger.debug("Retry failed with BulkOpPartialResultException: {} Before retry: {}", pre,
              result.getKeyListString());
          result.consolidate(pre.getResult());
        } catch (Exception rte) {
          oneSubMapRetryFailed = true;
          Object firstKey = newKeys.iterator().next();
          result.saveFailedKey(firstKey, rte);
        }
      } // for failedServer

      // If all retries succeeded, the PRE in first tries can be ignored
      if (oneSubMapRetryFailed && result.hasFailure()) {
        PutAllPartialResultException pre = new PutAllPartialResultException(result);
        throw pre;
      }
    } // failedServers!=null

    return result.getSucceededKeysAndVersions();
  }
Does a region put on a server using connections from the given pool to communicate with the
server.

@param pool the pool to use to communicate with the server.
@param region the name of the region to do the removeAll on
@param keys the Collection of keys to remove
@param eventId the event id for this removeAll
geode-core/src/main/java/org/apache/geode/cache/client/internal/RemoveAllOp.java