@Override
  protected boolean operateOnPartitionedRegion(final ClusterDistributionManager dm,
      PartitionedRegion r, long startTime) throws ForceReattemptException {
    if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
      logger.trace(LogMarker.DM_VERBOSE, "GetMessage operateOnRegion: {}", r.getFullPath());
    }

    PartitionedRegionDataStore ds = r.getDataStore();

    assert getTXUniqId() == TXManagerImpl.NOTX || r.getDataView() instanceof TXStateProxy;

    RawValue valueBytes;
    Object val = null;
    try {
      if (ds != null) {
        VersionTagHolder event = new VersionTagHolder();
        try {
          KeyInfo keyInfo = r.getKeyInfo(key, cbArg);
          boolean lockEntry = forceUseOfPRExecutor || isDirectAck();

          val = r.getDataView().getSerializedValue(r, keyInfo, !lockEntry, context, event,
              returnTombstones);

          if (val == BucketRegion.REQUIRES_ENTRY_LOCK) {
            Assert.assertTrue(!lockEntry);
            forceUseOfPRExecutor = true;
            if (logger.isDebugEnabled()) {
              logger.debug("Rescheduling GetMessage due to possible cache-miss");
            }
            schedule(dm);
            return false;
          }
          valueBytes = val instanceof RawValue ? (RawValue) val : new RawValue(val);
        } catch (DistributedSystemDisconnectedException sde) {
          sendReply(getSender(), processorId, dm,
              new ReplyException(new ForceReattemptException(
                  "Operation got interrupted due to shutdown in progress on remote VM.",
                  sde)),
              r, startTime);
          return false;
        } catch (PrimaryBucketException | DataLocationException pbe) {
          sendReply(getSender(), getProcessorId(), dm, new ReplyException(pbe), r, startTime);
          return false;
        }

        if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
          logger.debug(LogMarker.DM_VERBOSE,
              "GetMessage sending serialized value {} back via GetReplyMessage using processorId: {}",
              valueBytes, getProcessorId());
        }

        r.getPrStats().endPartitionMessagesProcessing(startTime);
        GetReplyMessage.send(getSender(), getProcessorId(), valueBytes, getReplySender(dm),
            event.getVersionTag());
        // Unless there was an exception thrown, this message handles sending the
        // response
        return false;
      } else {
        throw new InternalGemFireError(
            "Get message sent to wrong member");
      }
    } finally {
      OffHeapHelper.release(val);
    }


  }
Empty constructor to satisfy {@link DataSerializer} requirements
geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/GetMessage.java