protected EntrySnapshot getEntryInBucket(final DistributedMember targetNode, final int bucketId,
      final Object key, boolean access, final boolean allowTombstones) {
    final int retryAttempts = calcRetry();
    if (logger.isTraceEnabled()) {
      logger.trace("getEntryInBucket: " + "Key key={} ({}) from: {} bucketId={}", key,
          key.hashCode(), targetNode, bucketStringForLogs(bucketId));
    }
    EntrySnapshot ret = null;
    int count = 0;
    RetryTimeKeeper retryTime = null;
    InternalDistributedMember retryNode = (InternalDistributedMember) targetNode;
    while (count <= retryAttempts) {
      // Every continuation should check for DM cancellation
      if (retryNode == null) {
        checkReadiness();
        if (retryTime == null) {
          retryTime = new RetryTimeKeeper(retryTimeout);
        }
        if (retryTime.overMaximum()) {
          break;
        }
        retryNode = getOrCreateNodeForBucketRead(bucketId);

        // No storage found for bucket, early out preventing hot loop, bug 36819
        if (retryNode == null) {
          checkShutdown();
          return null;
        }
        continue;
      }
      try {
        final boolean loc = (localMaxMemory > 0) && retryNode.equals(getMyId());
        if (loc) {
          ret = dataStore.getEntryLocally(bucketId, key, access, allowTombstones);
        } else {
          ret = getEntryRemotely(retryNode, bucketId, key, access, allowTombstones);
          // TODO:Suranjan&Yogesh : there should be better way than this one
          String name = Thread.currentThread().getName();
          if (name.startsWith("ServerConnection") && !getMyId().equals(targetNode)) {
            setNetworkHopType(bucketId, (InternalDistributedMember) targetNode);
          }
        }

        return ret;
      } catch (PRLocallyDestroyedException pde) {
        if (logger.isDebugEnabled()) {
          logger.debug("getEntryInBucket: Encountered PRLocallyDestroyedException", pde);
        }
        checkReadiness();
      } catch (EntryNotFoundException ignore) {
        return null;
      } catch (ForceReattemptException prce) {
        prce.checkKey(key);
        if (logger.isDebugEnabled()) {
          logger.debug("getEntryInBucket: retrying, attempts so far: {}", count, prce);
        }
        checkReadiness();
        InternalDistributedMember lastNode = retryNode;
        retryNode = getOrCreateNodeForBucketRead(bucketId);
        if (lastNode.equals(retryNode)) {
          if (retryTime == null) {
            retryTime = new RetryTimeKeeper(retryTimeout);
          }
          if (retryTime.overMaximum()) {
            break;
          }
          retryTime.waitToRetryNode();
        }
      } catch (PrimaryBucketException notPrimary) {
        if (logger.isDebugEnabled()) {
          logger.debug("Bucket {} on Node {} not primary", notPrimary.getLocalizedMessage(),
              retryNode);
        }
        getRegionAdvisor().notPrimary(bucketId, retryNode);
        retryNode = getOrCreateNodeForBucketRead(bucketId);
      }

      // It's possible this is a GemFire thread e.g. ServerConnection
      // which got to this point because of a distributed system shutdown or
      // region closure which uses interrupt to break any sleep() or wait()
      // calls
      // e.g. waitForPrimary
      checkShutdown();

      count++;
      if (count == 1) {
        prStats.incContainsKeyValueOpsRetried();
      }
      prStats.incContainsKeyValueRetries();

    }

    PartitionedRegionDistributionException e = null; // Fix for bug 36014
    if (logger.isDebugEnabled()) {
      e = new PartitionedRegionDistributionException(
          String.format("No VM available for getEntry in %s attempts",
              count));
    }
    logger.warn(String.format("No VM available for getEntry in %s attempts", count), e);
    return null;
  }
@param access true if caller wants last accessed time updated
@param allowTombstones - whether a tombstone can be returned
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java