private Object getFromBucket(final InternalDistributedMember targetNode, int bucketId,
      final Object key, final Object aCallbackArgument, boolean disableCopyOnRead, boolean preferCD,
      ClientProxyMembershipID requestingClient, EntryEventImpl clientEvent,
      boolean returnTombstones, boolean allowRetry) {
    final boolean isDebugEnabled = logger.isDebugEnabled();

    final int retryAttempts = calcRetry();
    Object obj;
    // retry the get remotely until it finds the right node managing the bucket
    int count = 0;
    RetryTimeKeeper retryTime = null;
    InternalDistributedMember retryNode = targetNode;
    while (count <= retryAttempts) {
      // Every continuation should check for DM cancellation
      if (retryNode == null) {
        checkReadiness();
        if (retryTime == null) {
          retryTime = new RetryTimeKeeper(retryTimeout);
        }
        retryNode = getNodeForBucketReadOrLoad(bucketId);

        // No storage found for bucket, early out preventing hot loop, bug 36819
        if (retryNode == null) {
          checkShutdown();
          return null;
        }
        continue;
      }
      final boolean isLocal = localMaxMemory > 0 && retryNode.equals(getMyId());

      try {
        if (isLocal) {
          obj = dataStore.getLocally(bucketId, key, aCallbackArgument, disableCopyOnRead,
              preferCD, requestingClient, clientEvent, returnTombstones, false);
        } else {
          if (haveCacheLoader) {
            // If the region has a cache loader,
            // the target node is the primary server of the bucket. But, if the
            // value can be found in a local bucket, we should first try there.

            /* MergeGemXDHDFSToGFE -readoing from local bucket was disabled in GemXD */
            if (null != (obj = getFromLocalBucket(bucketId, key, aCallbackArgument,
                disableCopyOnRead, preferCD, requestingClient, clientEvent, returnTombstones))) {
              return obj;
            }
          }

          obj = getRemotely(retryNode, bucketId, key, aCallbackArgument, preferCD, requestingClient,
              clientEvent, returnTombstones);

          // TODO: there should be better way than this one
          String name = Thread.currentThread().getName();
          if (name.startsWith("ServerConnection") && !getMyId().equals(retryNode)) {
            setNetworkHopType(bucketId, retryNode);
          }
        }
        return obj;
      } catch (PRLocallyDestroyedException pde) {
        if (isDebugEnabled) {
          logger.debug("getFromBucket Encountered PRLocallyDestroyedException", pde);
        }
        checkReadiness();
        if (allowRetry) {
          retryNode = getNodeForBucketReadOrLoad(bucketId);
        } else {
          // Only transactions set allowRetry to false,
          // fail the transaction here as region is destroyed.
          Throwable cause = pde.getCause();
          if (cause != null && cause instanceof RegionDestroyedException) {
            throw (RegionDestroyedException) cause;
          } else {
            // Should not see it currently, all current constructors of PRLocallyDestroyedException
            // set the cause to RegionDestroyedException.
            throw new RegionDestroyedException(toString(), getFullPath());
          }
        }
      } catch (ForceReattemptException prce) {
        prce.checkKey(key);
        checkReadiness();

        if (allowRetry) {
          InternalDistributedMember lastNode = retryNode;
          if (isDebugEnabled) {
            logger.debug("getFromBucket: retry attempt: {} of {}", count, retryAttempts, prce);
          }
          retryNode = getNodeForBucketReadOrLoad(bucketId);
          if (lastNode.equals(retryNode)) {
            if (retryTime == null) {
              retryTime = new RetryTimeKeeper(retryTimeout);
            }
            if (retryTime.overMaximum()) {
              break;
            }
            if (isDebugEnabled) {
              logger.debug("waiting to retry node {}", retryNode);
            }
            retryTime.waitToRetryNode();
          }
        } else {
          // with transaction
          handleForceReattemptExceptionWithTransaction(prce);
        }
      } catch (PrimaryBucketException notPrimary) {
        if (allowRetry) {
          if (isDebugEnabled) {
            logger.debug("getFromBucket: {} on Node {} not primary",
                notPrimary.getLocalizedMessage(), retryNode);
          }
          getRegionAdvisor().notPrimary(bucketId, retryNode);
          retryNode = getNodeForBucketReadOrLoad(bucketId);
        } else {
          throw notPrimary;
        }
      }

      // 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.incGetOpsRetried();
      }
      prStats.incGetRetries();
      if (isDebugEnabled) {
        logger.debug("getFromBucket: Attempting to resend get to node {} after {} failed attempts",
            retryNode, count);
      }
    } // While

    PartitionedRegionDistributionException e = null; // Fix for bug 36014
    if (logger.isDebugEnabled()) {
      e = new PartitionedRegionDistributionException(
          String.format("No VM available for get in %s attempts",
              count));
    }
    logger.warn(String.format("No VM available for get in %s attempts", count), e);
    return null;
  }
@param requestingClient the client requesting the object, or null if not from a client
@param allowRetry if false then do not retry
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java