private void _getKeysWithInterest(final @NotNull InterestType interestType, Object interestArg,
      boolean allowTombstones,
      SetCollector collector) throws IOException {
    // this could be parallelized by building up a list of buckets for each
    // vm and sending out the requests for keys in parallel. That might dump
    // more onto this vm in one swoop than it could handle, though, so we're
    // keeping it simple for now
    int totalBuckets = getTotalNumberOfBuckets();
    int retryAttempts = calcRetry();
    for (int bucket = 0; bucket < totalBuckets; bucket++) {
      Set bucketSet = null;
      final RetryTimeKeeper retryTime = new RetryTimeKeeper(Integer.MAX_VALUE);
      InternalDistributedMember bucketNode = getOrCreateNodeForBucketRead(bucket);
      for (int count = 0; count <= retryAttempts; count++) {
        if (logger.isDebugEnabled()) {
          logger.debug("_getKeysWithInterest bucketId={} attempt={}", bucket, (count + 1));
        }
        try {
          if (bucketNode != null) {
            if (bucketNode.equals(getMyId())) {
              bucketSet = dataStore.handleRemoteGetKeys(bucket, interestType, interestArg,
                  allowTombstones);
            } else {
              FetchKeysResponse r = FetchKeysMessage.sendInterestQuery(bucketNode, this, bucket,
                  interestType, interestArg, allowTombstones);
              bucketSet = r.waitForKeys();
            }
          }
          break;
        } catch (PRLocallyDestroyedException ignore) {
          if (logger.isDebugEnabled()) {
            logger.debug("_getKeysWithInterest: Encountered PRLocallyDestroyedException");
          }
          checkReadiness();
        } catch (ForceReattemptException prce) {
          // no checkKey possible
          if (logger.isDebugEnabled()) {
            logger.debug("_getKeysWithInterest: retry attempt: {}", count, prce);
          }
          checkReadiness();

          InternalDistributedMember lastTarget = bucketNode;
          bucketNode = getOrCreateNodeForBucketRead(bucket);
          if (lastTarget.equals(bucketNode)) {
            if (retryTime.overMaximum()) {
              break;
            }
            retryTime.waitToRetryNode();
          }
        }
      } // for(count)
      if (bucketSet != null) {
        collector.receiveSet(bucketSet);
      }
    } // for(bucket)
  }
finds all the keys matching the given interest type and passes them to the given collector

@param allowTombstones whether to return destroyed entries
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java