VersionedObjectList basicPutAll(final Map<?, ?> map,
      final DistributedPutAllOperation putAllOp, final Map<Object, VersionTag> retryVersions) {

    final boolean isDebugEnabled = logger.isDebugEnabled();
    final EntryEventImpl event = putAllOp.getBaseEvent();
    EventID eventId = event.getEventId();
    if (eventId == null && generateEventID()) {
      // We need to "reserve" the eventIds for the entries in map here
      event.reserveNewEventId(cache.getDistributedSystem(), map.size());
      eventId = event.getEventId();
    }

    verifyPutAllMap(map);
    VersionedObjectList proxyResult = null;
    boolean partialResult = false;
    RuntimeException runtimeException = null;
    if (hasServerProxy()) {
      // send message to cache server
      if (isTX()) {
        TXStateProxyImpl txState = (TXStateProxyImpl) cache.getTxManager().getTXState();
        txState.getRealDeal(null, this);
      }
      try {
        proxyResult = getServerProxy().putAll(
            uncheckedCast(map),
            eventId,
            !event.isGenerateCallbacks(),
            event.getCallbackArgument());
        if (isDebugEnabled) {
          logger.debug("PutAll received response from server: {}", proxyResult);
        }
      } catch (PutAllPartialResultException e) {
        // adjust the map to only add succeeded entries, then apply the adjustedMap
        proxyResult = e.getSucceededKeysAndVersions();
        partialResult = true;
        if (isDebugEnabled) {
          logger.debug(
              "putAll in client encountered a PutAllPartialResultException:{}{}. Adjusted keys are: {}",
              e.getMessage(), lineSeparator(), proxyResult.getKeys());
        }
        Throwable txException = e.getFailure();
        while (txException != null) {
          if (txException instanceof TransactionException) {
            runtimeException = (RuntimeException) txException;
            break;
          }
          txException = txException.getCause();
        }
        if (runtimeException == null) {
          // for cache close
          runtimeException = getCancelCriterion().generateCancelledException(e.getFailure());
          if (runtimeException == null) {
            runtimeException = new ServerOperationException(
                String.format("Region %s putAll at server applied partial keys due to exception.",
                    getFullPath()),
                e.getFailure());
          }
        }
      }
    }

    final VersionedObjectList succeeded =
        new VersionedObjectList(map.size(), true, getConcurrencyChecksEnabled());

    // if this is a transactional putAll, we will not have version information as it is only
    // generated at commit
    // so treat transactional putAll as if the server is not versioned
    final boolean serverIsVersioned = proxyResult != null && proxyResult.regionIsVersioned()
        && !isTX() && getDataPolicy() != DataPolicy.EMPTY;
    if (!serverIsVersioned && !partialResult) {
      // we don't need server information if it isn't versioned or if the region is empty
      proxyResult = null;
    }

    lockRVVForBulkOp();
    try {
      try {
        int size = proxyResult == null ? map.size() : proxyResult.size();

        if (isDebugEnabled) {
          logger.debug("size of put result is {} maps is {} proxyResult is {}", size, map,
              proxyResult);
        }

        final PutAllPartialResult partialKeys = new PutAllPartialResult(size);
        final Iterator iterator;
        final boolean isVersionedResults;
        if (proxyResult != null) {
          iterator = proxyResult.iterator();
          isVersionedResults = true;
        } else {
          iterator = map.entrySet().iterator();
          isVersionedResults = false;
        }

        // TODO: refactor this mess
        Runnable task = () -> {
          int offset = 0;
          VersionTagHolder tagHolder = new VersionTagHolder();
          while (iterator.hasNext()) {
            stopper.checkCancelInProgress(null);
            Map.Entry mapEntry = (Map.Entry) iterator.next();
            Object key = mapEntry.getKey();
            tagHolder.setVersionTag(null);
            final Object value;
            boolean overwritten = false;
            VersionTag versionTag = null;
            if (isVersionedResults) {
              versionTag = ((VersionedObjectList.Entry) mapEntry).getVersionTag();
              value = map.get(key);
              if (isDebugEnabled) {
                logger.debug("putAll key {} -> {} version={}", key, value, versionTag);
              }
              if (versionTag == null && serverIsVersioned && getConcurrencyChecksEnabled()
                  && getDataPolicy().withStorage()) {
                // server was unable to determine the version for this operation.
                // I'm not sure this can still happen as described below on a pr.
                // But it can happen on the server if NORMAL or PRELOADED.
                // This can happen in a PR with redundancy if there is a bucket
                // failure or migration during the operation. We destroy the
                // entry since we don't know what its state should be (but the server should)
                if (isDebugEnabled) {
                  logger.debug("server returned no version information for {}", key);
                }
                localDestroyNoCallbacks(key);
                // to be consistent we need to fetch the current entry
                get(key, event.getCallbackArgument(), false, null);
                overwritten = true;
              }
            } else {
              value = mapEntry.getValue();
              if (isDebugEnabled) {
                logger.debug("putAll {} -> {}", key, value);
              }
            }
            try {
              if (serverIsVersioned) {
                if (isDebugEnabled) {
                  logger.debug("associating version tag with {} version={}", key, versionTag);
                }
                // If we have received a version tag from a server, add it to the event
                tagHolder.setVersionTag(versionTag);
                tagHolder.setFromServer(true);
              } else if (retryVersions != null && retryVersions.containsKey(key)) {
                // If this is a retried event, and we have a version tag for the retry,
                // add it to the event.
                tagHolder.setVersionTag(retryVersions.get(key));
              }

              if (!overwritten) {
                basicEntryPutAll(key, value, putAllOp, offset, tagHolder);
              }
              // now we must check again since the cache may have closed during
              // distribution (causing this process to not receive and queue the
              // event for clients
              stopper.checkCancelInProgress(null);
              succeeded.addKeyAndVersion(key, tagHolder.getVersionTag());
            } catch (Exception ex) {
              if (isDebugEnabled) {
                logger.debug("PutAll operation encountered exception for key {}", key, ex);
              }
              partialKeys.saveFailedKey(key, ex);
            }
            offset++;
          }
        };

        syncBulkOp(task, eventId);
        if (partialKeys.hasFailure()) {
          // Now succeeded contains an order key list, may be missing the version tags.
          // Save reference of succeeded into partialKeys. The succeeded may be modified by
          // postPutAll() to fill in the version tags.
          partialKeys.setSucceededKeysAndVersions(succeeded);
          logger
              .info("Region {} putAll: {}",
                  new Object[] {getFullPath(), partialKeys});
          if (isDebugEnabled) {
            logger.debug(partialKeys.detailString());
          }
          if (runtimeException == null) {
            // if received exception from server first, ignore local exception
            if (putAllOp.isBridgeOperation()) {
              if (partialKeys.getFailure() instanceof CancelException) {
                runtimeException = (RuntimeException) partialKeys.getFailure();
              } else if (partialKeys.getFailure() instanceof LowMemoryException) {
                throw partialKeys.getFailure();
              } else {
                runtimeException = new PutAllPartialResultException(partialKeys);
                if (isDebugEnabled) {
                  logger.debug("basicPutAll:" + partialKeys.detailString());
                }
              }
            } else {
              throw partialKeys.getFailure();
            }
          }
        }
      } catch (LowMemoryException lme) {
        throw lme;
      } catch (RuntimeException ex) {
        runtimeException = ex;
      } catch (Exception ex) {
        runtimeException = new RuntimeException(ex);
      } finally {
        putAllOp.getBaseEvent().release();
        putAllOp.freeOffHeapResources();
      }
      getDataView().postPutAll(putAllOp, succeeded, this);
    } finally {
      unlockRVVForBulkOp();
    }
    if (runtimeException != null) {
      throw runtimeException;
    }
    return succeeded;
  }
Returns true if a one-hop (RemoteOperationMessage) should be used when applying the change to
the system.
geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java