@Override
  protected boolean operateOnPartitionedRegion(ClusterDistributionManager dm, PartitionedRegion r,
      long startTime) throws CacheException, ForceReattemptException {
    if (logger.isDebugEnabled()) {
      logger.debug("FetchKeysMessage operateOnRegion: {} bucketId: {} type: {} {}", r.getFullPath(),
          bucketId, InterestType.getString(interestType),
          (allowTombstones ? " with tombstones" : " without tombstones"));
    }

    PartitionedRegionDataStore ds = r.getDataStore();
    if (ds != null) {
      try {
        Set<?> keys =
            ds.handleRemoteGetKeys(bucketId, interestType, interestArg, allowTombstones);
        if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
          logger.trace(LogMarker.DM_VERBOSE,
              "FetchKeysMessage sending {} keys back using processorId: {} : {}", keys.size(),
              getProcessorId(), keys);
        }
        r.getPrStats().endPartitionMessagesProcessing(startTime);
        FetchKeysReplyMessage.send(getSender(), getProcessorId(), dm, keys);
      } catch (PRLocallyDestroyedException pde) {
        if (logger.isDebugEnabled()) {
          logger.debug("FetchKeysMessage Encountered PRLocallyDestroyedException");
        }
        throw new ForceReattemptException(
            "Encountered PRLocallyDestroyedException",
            pde);
      }
    } else {
      logger.warn("FetchKeysMessage: data store not configured for this member");
    }

    // Unless there was an exception thrown, this message handles sending the response
    return false;
  }
@return the FetchKeysResponse
@throws ForceReattemptException if the peer is no longer available
geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/FetchKeysMessage.java