public static FetchKeysResponse sendInterestQuery(InternalDistributedMember recipient,
      PartitionedRegion r, Integer bucketId, final @NotNull InterestType interestType, Object arg,
      boolean allowTombstones)
      throws ForceReattemptException {
    Assert.assertTrue(recipient != null, "FetchKeysMessage NULL recipient");
    FetchKeysMessage tmp = new FetchKeysMessage();
    FetchKeysResponse p =
        (FetchKeysResponse) tmp.createReplyProcessor(r, Collections.singleton(recipient));
    FetchKeysMessage m =
        new FetchKeysMessage(recipient, r.getPRId(), p, bucketId, interestType, arg,
            allowTombstones);
    m.setTransactionDistributed(r.getCache().getTxManager().isDistributed());
    Set<InternalDistributedMember> failures = r.getDistributionManager().putOutgoing(m);
    if (failures != null && failures.size() > 0) {
      throw new ForceReattemptException(
          String.format("Failed sending < %s >", m));
    }

    return p;
  }
@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