void doGet(ClusterDistributionManager dm) {
      long startTime = dm.cacheTimeMillis();
      // boolean retVal = true;
      byte[] ebv = null;
      Object ebvObj = null;
      int ebvLen = 0;
      long lastModifiedCacheTime = 0;
      boolean isSer = false;
      boolean requestorTimedOut = false;
      boolean authoritative = false;
      VersionTag versionTag = null;

      InternalCache cache = dm.getExistingCache();

      final InitializationLevel oldLevel =
          LocalRegion.setThreadInitLevelRequirement(BEFORE_INITIAL_IMAGE);
      LocalRegion region = (LocalRegion) cache.getRegion(regionName);
      CachePerfStats stats =
          region == null ? cache.getCachePerfStats() : region.getRegionPerfStats();
      long startHandlingTime = stats.startHandlingNetsearch();
      boolean handlingSuccess = false;
      try {
        if (region != null) {
          setClearCountReference(region);
          try {
            boolean initialized = region.isInitialized();
            RegionEntry entry = region.basicGetEntry(key);
            if (entry != null) {
              synchronized (entry) {
                // get the value and version under synchronization so they don't change
                VersionStamp versionStamp = entry.getVersionStamp();
                if (versionStamp != null) {
                  versionTag = versionStamp.asVersionTag();
                }
                Object eov = region.getNoLRU(key, false, true, true); // OFFHEAP: incrc, copy
                // bytes, decrc
                if (eov != null) {
                  if (eov == Token.INVALID || eov == Token.LOCAL_INVALID) {
                    // nothing?
                  } else if (dm.cacheTimeMillis() - startTime < timeoutMs) {
                    if (!region.isExpiredWithRegardTo(key, ttl, idleTime)) {
                      lastModifiedCacheTime = entry.getLastModified();
                      if (eov instanceof CachedDeserializable) {
                        CachedDeserializable cd = (CachedDeserializable) eov;
                        if (!cd.isSerialized()) {
                          isSer = false;
                          ebv = (byte[]) cd.getDeserializedForReading();
                          ebvLen = ebv.length;
                        } else {
                          // don't serialize here if it is not already serialized
                          Object tmp = cd.getValue();
                          if (tmp instanceof byte[]) {
                            byte[] bb = (byte[]) tmp;
                            ebv = bb;
                            ebvLen = bb.length;
                          } else {
                            ebvObj = tmp;
                          }
                          isSer = true;
                        }
                      } else if (!(eov instanceof byte[])) {
                        ebvObj = eov;
                        isSer = true;
                      } else {
                        ebv = (byte[]) eov;
                        ebvLen = ebv.length;
                      }
                    }
                    handlingSuccess = true;
                  } else {
                    requestorTimedOut = true;
                  }
                }
              }
            }
            authoritative =
                region.getDataPolicy().withReplication() && initialized && !region.isDestroyed;
          } finally {
            removeClearCountReference(region);
          }
        }
        NetSearchReplyMessage.sendMessage(getSender(), processorId,
            key, ebv, ebvObj, ebvLen, lastModifiedCacheTime, isSer, requestorTimedOut,
            authoritative, dm, versionTag);
      } catch (RegionDestroyedException ignore) {
        replyWithNull(dm);

      } catch (CancelException ignore) {
        replyWithNull(dm);

      } catch (VirtualMachineError err) {
        SystemFailure.initiateFailure(err);
        // If this ever returns, rethrow the error. We're poisoned
        // now, so don't let this thread continue.
        throw err;
      } catch (Throwable t) {
        // Whenever you catch Error or Throwable, you must also
        // catch VirtualMachineError (see above). However, there is
        // _still_ a possibility that you are dealing with a cascading
        // error condition, so you also need to check to see if the JVM
        // is still usable:
        SystemFailure.checkFailure();
        logger.warn("Unexpected exception creating net search reply", t);
        replyWithNull(dm);
      } finally {
        LocalRegion.setThreadInitLevelRequirement(oldLevel);
        stats.endHandlingNetsearch(startHandlingTime, handlingSuccess);
      }
    }
Invoked on the node that has the object
geode-core/src/main/java/org/apache/geode/internal/cache/SearchLoadAndWriteProcessor.java