public boolean executeQueryOnRemoteAndLocalNodes(final TestHook th)
      throws InterruptedException, QueryException {
    final boolean isDebugEnabled = logger.isDebugEnabled();

    if (Thread.interrupted()) {
      throw new InterruptedException();
    }

    HashMap<InternalDistributedMember, List<Integer>> n2b =
        new HashMap<>(node2bucketIds);
    n2b.remove(pr.getMyId());
    // Shobhit: IF query is originated from a Function and we found some buckets on
    // remote node we should throw exception mentioning data movement during function execution.
    // According to discussions we dont know if this is possible as buckets are not moved until
    // function execution is completed.
    if (query.isQueryWithFunctionContext() && !n2b.isEmpty()) {
      if (isDebugEnabled) {
        logger.debug("Remote buckets found for query executed in a Function.");
      }
      throw new QueryInvocationTargetException(
          "Data movement detected accross PartitionRegion nodes while executing the Query with function filter.");
    }

    if (isDebugEnabled) {
      logger.debug("Sending query execution request to {} remote members for the query:{}",
          n2b.size(), query.getQueryString());
    }
    StreamingQueryPartitionResponse processor = null;
    boolean requiresRetry = false;

    if (n2b.isEmpty()) {
      if (isDebugEnabled) {
        logger.debug("No remote members with buckets to query.");
      }
    } else {
      // send separate message to each recipient since each one has a
      // different list of bucket ids
      processor = createStreamingQueryPartitionResponse(sys, n2b);
      for (Map.Entry<InternalDistributedMember, List<Integer>> me : n2b.entrySet()) {
        final InternalDistributedMember rcp = me.getKey();
        final List<Integer> bucketIds = me.getValue();
        PartitionMessage m = createRequestMessage(rcp, processor, bucketIds);
        m.setTransactionDistributed(sys.getCache().getTxManager().isDistributed());
        Set notReceivedMembers = sendMessage(m);
        if (th != null) {
          th.hook(4);
        }
        if (notReceivedMembers != null && !notReceivedMembers.isEmpty()) {
          requiresRetry = true;
          processor.removeFailedSenders(notReceivedMembers);
          if (isDebugEnabled) {
            logger.debug("Failed sending to members {} retry required", notReceivedMembers);
          }
        }
      }
      if (th != null) {
        th.hook(5);
      }

    }

    Throwable localFault = null;
    boolean localNeedsRetry = false;

    // Shobhit: Check if query is only for local buckets else return.
    if (node2bucketIds.containsKey(pr.getMyId())) {
      if (isDebugEnabled) {
        logger.debug("Started query execution on local data for query:{}",
            query.getQueryString());
      }

      try {
        localNeedsRetry = executeQueryOnLocalNode();
        if (th != null) {
          th.hook(0);
        }
      } catch (VirtualMachineError e) {
        SystemFailure.initiateFailure(e);
        throw e;
      } catch (Throwable t) {
        SystemFailure.checkFailure();
        // We have to wait for remote exceptions
        // if request was sent to remote nodes.
        localFault = t;
      }
    } else {
      if (isDebugEnabled) {
        logger.debug("No local buckets to query.");
      }
    }

    if (processor != null) {
      try {
        // should we allow this to timeout?
        failedMembers = processor.waitForCacheOrQueryException();
        for (InternalDistributedMember member : failedMembers) {
          memberStreamCorrupted(member);
        }
        requiresRetry |= !failedMembers.isEmpty();

        if (isDebugEnabled) {
          logger.debug("Following remote members failed {} and retry flag is set to: {}",
              failedMembers, requiresRetry);
        }
      } catch (org.apache.geode.cache.TimeoutException e) { // Shobhit: Swallow remote exception if
                                                            // local exception is there.
        if (localFault == null) {
          throw new QueryException(e);
        }
      } catch (ReplyException e) {
        if (localFault == null) {
          throw e;
        }
      } catch (Error e) {
        if (localFault == null) {
          throw e;
        }
      } catch (RuntimeException e) {
        if (localFault == null) {
          throw e;
        }
      }
    }

    if (executionContext.isCanceled()) {
      throw executionContext.getQueryCanceledException();
    }

    if (localFault != null) {
      if (localFault instanceof QueryException) {
        throw (QueryException) localFault;
      } else if (localFault instanceof InterruptedException) {
        throw (InterruptedException) localFault;
      } else if (localFault instanceof Error) {
        throw (Error) localFault;
      } else if (localFault instanceof RuntimeException) {
        throw (RuntimeException) localFault;
      }
    }
    return requiresRetry | localNeedsRetry;
  }
Returns normally if succeeded to get data, otherwise throws an exception

@param th a test hook
@return true if parts of the query need to be retried, otherwise false
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionQueryEvaluator.java