@Override
  public long postPutAllSend(DistributedPutAllOperation putAllOp,
      VersionedObjectList successfulPuts) {
    final boolean isDebugEnabled = logger.isDebugEnabled();

    if (cache.isCacheAtShutdownAll()) {
      throw cache.getCacheClosedException("Cache is shutting down");
    }

    final long startTime = prStats.getTime();
    // build all the msgs by bucketid
    HashMap prMsgMap = putAllOp.createPRMessages();
    PutAllPartialResult partialKeys = new PutAllPartialResult(putAllOp.putAllDataSize);

    // clear the successfulPuts list since we're actually doing the puts here
    // and the basicPutAll work was just a way to build the DPAO object
    Map<Object, VersionTag> keyToVersionMap =
        new HashMap<>(successfulPuts.size());
    successfulPuts.clearVersions();
    for (final Object value : prMsgMap.entrySet()) {
      Map.Entry mapEntry = (Map.Entry) value;
      Integer bucketId = (Integer) mapEntry.getKey();
      PutAllPRMessage prMsg = (PutAllPRMessage) mapEntry.getValue();
      checkReadiness();
      long then = 0;
      if (isDebugEnabled) {
        then = System.currentTimeMillis();
      }
      try {
        VersionedObjectList versions = sendMsgByBucket(bucketId, prMsg);
        if (versions.size() > 0) {
          partialKeys.addKeysAndVersions(versions);
          versions.saveVersions(keyToVersionMap);
        } else if (!getConcurrencyChecksEnabled()) { // no keys returned if not versioned
          Set keys = prMsg.getKeys();
          partialKeys.addKeys(keys);
        }
      } catch (PutAllPartialResultException pre) {
        // sendMsgByBucket applied partial keys
        if (isDebugEnabled) {
          logger.debug("PR.postPutAll encountered PutAllPartialResultException, ", pre);
        }
        partialKeys.consolidate(pre.getResult());
      } catch (Exception ex) {
        // If failed at other exception
        if (isDebugEnabled) {
          logger.debug("PR.postPutAll encountered exception at sendMsgByBucket, ", ex);
        }
        @Released
        EntryEventImpl firstEvent = prMsg.getFirstEvent(this);
        try {
          partialKeys.saveFailedKey(firstEvent.getKey(), ex);
        } finally {
          firstEvent.release();
        }
      }
      if (isDebugEnabled) {
        long now = System.currentTimeMillis();
        if ((now - then) >= 10000) {
          logger.debug("PR.sendMsgByBucket took " + (now - then) + " ms");
        }
      }
    }
    prStats.endPutAll(startTime);
    if (!keyToVersionMap.isEmpty()) {
      for (final Object o : successfulPuts.getKeys()) {
        successfulPuts.addVersion(keyToVersionMap.get(o));
      }
      keyToVersionMap.clear();
    }

    if (partialKeys.hasFailure()) {
      logger.info("Region {} putAll: {}",
          new Object[] {getFullPath(), partialKeys});
      if (putAllOp.isBridgeOperation()) {
        if (partialKeys.getFailure() instanceof CancelException) {
          throw (CancelException) partialKeys.getFailure();
        } else {
          throw new PutAllPartialResultException(partialKeys);
        }
      } else {
        if (partialKeys.getFailure() instanceof RuntimeException) {
          throw (RuntimeException) partialKeys.getFailure();
        } else {
          throw new RuntimeException(partialKeys.getFailure());
        }
      }
    }
    return -1;
  }
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