private VersionedObjectList sendMsgByBucket(final Integer bucketId, PutAllPRMessage prMsg) {
    final boolean isDebugEnabled = logger.isDebugEnabled();

    // retry the put remotely until it finds the right node managing the bucket
    @Released
    EntryEventImpl event = prMsg.getFirstEvent(this);
    try {
      RetryTimeKeeper retryTime = null;
      InternalDistributedMember currentTarget = getNodeForBucketWrite(bucketId, null);
      if (isDebugEnabled) {
        logger.debug("PR.sendMsgByBucket:bucket {}'s currentTarget is {}", bucketId, currentTarget);
      }

      long timeOut = 0;
      int count = 0;
      for (;;) {
        switch (count) {
          case 0:
            // Note we don't check for DM cancellation in common case.
            // First time. Assume success, keep going.
            break;
          case 1:
            cache.getCancelCriterion().checkCancelInProgress(null);
            // Second time (first failure). Calculate timeout and keep going.
            timeOut = System.currentTimeMillis() + retryTimeout;
            break;
          default:
            cache.getCancelCriterion().checkCancelInProgress(null);
            // test for timeout
            long timeLeft = timeOut - System.currentTimeMillis();
            if (timeLeft < 0) {
              PRHARedundancyProvider.timedOut(this, null, null, "update an entry",
                  retryTimeout);
              // NOTREACHED
            }

            // Didn't time out. Sleep a bit and then continue
            boolean interrupted = Thread.interrupted();
            try {
              Thread.sleep(PartitionedRegionHelper.DEFAULT_WAIT_PER_RETRY_ITERATION);
            } catch (InterruptedException ignore) {
              interrupted = true;
            } finally {
              if (interrupted) {
                Thread.currentThread().interrupt();
              }
            }
            break;
        } // switch
        count++;

        if (currentTarget == null) { // pick target
          checkReadiness();
          if (retryTime == null) {
            retryTime = new RetryTimeKeeper(retryTimeout);
          }

          currentTarget = waitForNodeOrCreateBucket(retryTime, event, bucketId);
          if (isDebugEnabled) {
            logger.debug("PR.sendMsgByBucket: event size is {}, new currentTarget is {}",
                getEntrySize(event), currentTarget);
          }

          // 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 or waitForBucketRecovery in which case throw exception
          checkShutdown();
          continue;
        } // pick target

        try {
          return tryToSendOnePutAllMessage(prMsg, currentTarget);
        } catch (ForceReattemptException prce) {
          checkReadiness();
          InternalDistributedMember lastTarget = currentTarget;
          if (retryTime == null) {
            retryTime = new RetryTimeKeeper(retryTimeout);
          }
          currentTarget = getNodeForBucketWrite(bucketId, retryTime);
          if (isDebugEnabled) {
            logger.debug("PR.sendMsgByBucket: Old target was {}, Retrying {}", lastTarget,
                currentTarget);
          }
          if (lastTarget.equals(currentTarget)) {
            if (isDebugEnabled) {
              logger.debug("PR.sendMsgByBucket: Retrying at the same node:{} due to {}",
                  currentTarget, prce.getMessage());
            }
            if (retryTime.overMaximum()) {
              PRHARedundancyProvider.timedOut(this, null, null, "update an entry",
                  retryTimeout);
              // NOTREACHED
            }
            retryTime.waitToRetryNode();
          }
          event.setPossibleDuplicate(true);
          prMsg.setPossibleDuplicate(true);
        } catch (PrimaryBucketException notPrimary) {
          if (isDebugEnabled) {
            logger.debug("Bucket {} on Node {} not primnary", notPrimary.getLocalizedMessage(),
                currentTarget);
          }
          getRegionAdvisor().notPrimary(bucketId, currentTarget);
          if (retryTime == null) {
            retryTime = new RetryTimeKeeper(retryTimeout);
          }
          currentTarget = getNodeForBucketWrite(bucketId, retryTime);
        } catch (DataLocationException dle) {
          if (isDebugEnabled) {
            logger.debug("DataLocationException processing putAll", dle);
          }
          throw new TransactionException(dle);
        }

        // 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 or waitForBucketRecovery in which case throw
        // exception
        checkShutdown();

        // If we get here, the attempt failed...
        if (count == 1) {
          prStats.incPutAllMsgsRetried();
        }
        prStats.incPutAllRetries();
      } // for
    } finally {
      if (event != null) {
        event.release();
      }
    }
    // NOTREACHED
  }
Create PutAllPRMsgs for each bucket, and send them.

@param putAllOp DistributedPutAllOperation object.
@param successfulPuts not used in PartitionedRegion.
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java