public boolean doLocalPutAll(final LocalRegion r, final InternalDistributedMember eventSender,
      long lastModified) throws EntryExistsException, RemoteOperationException {
    final DistributedRegion dr = (DistributedRegion) r;

    @Released
    EntryEventImpl baseEvent = EntryEventImpl.create(r, Operation.PUTALL_CREATE, null, null,
        callbackArg, false, eventSender, !skipCallbacks);
    try {

      baseEvent.setCausedByMessage(this);

      // set baseEventId to the first entry's event id. We need the thread id for DACE
      baseEvent.setEventId(eventId);
      if (bridgeContext != null) {
        baseEvent.setContext(bridgeContext);
      }
      baseEvent.setPossibleDuplicate(posDup);
      if (logger.isDebugEnabled()) {
        logger.debug(
            "RemotePutAllMessage.doLocalPutAll: eventSender is {}, baseEvent is {}, msg is {}",
            eventSender, baseEvent, this);
      }
      final DistributedPutAllOperation dpao =
          new DistributedPutAllOperation(baseEvent, putAllDataCount, false);
      try {
        r.lockRVVForBulkOp();
        final VersionedObjectList versions =
            new VersionedObjectList(putAllDataCount, true, dr.getConcurrencyChecksEnabled());
        dr.syncBulkOp(() -> {
          InternalDistributedMember myId = r.getDistributionManager().getDistributionManagerId();
          for (int i = 0; i < putAllDataCount; ++i) {
            @Released
            EntryEventImpl ev = PutAllPRMessage.getEventFromEntry(r, myId, eventSender, i,
                putAllData, false, bridgeContext, posDup, !skipCallbacks);
            try {
              ev.setPutAllOperation(dpao);
              if (logger.isDebugEnabled()) {
                logger.debug("invoking basicPut with {}", ev);
              }
              if (dr.basicPut(ev, false, false, null, false)) {
                putAllData[i].versionTag = ev.getVersionTag();
                versions.addKeyAndVersion(putAllData[i].getKey(), ev.getVersionTag());
              }
            } finally {
              ev.release();
            }
          }
        }, baseEvent.getEventId());
        if (getTXUniqId() != TXManagerImpl.NOTX || dr.getConcurrencyChecksEnabled()) {
          dr.getDataView().postPutAll(dpao, versions, dr);
        }
        PutAllReplyMessage.send(getSender(), processorId,
            getReplySender(r.getDistributionManager()), versions, putAllData,
            putAllDataCount);
        return false;
      } finally {
        r.unlockRVVForBulkOp();
        dpao.freeOffHeapResources();
      }
    } finally {
      baseEvent.release();
    }
  }
This method is called by both operateOnLocalRegion() when processing a remote msg or by
sendMsgByBucket() when processing a msg targeted to local Jvm. LocalRegion Note: It is very
important that this message does NOT cause any deadlocks as the sender will wait indefinitely
for the acknowledgment

@param r region
@param eventSender the endpoint server who received request from client
@param lastModified timestamp for last modification
@return If succeeds, return true, otherwise, throw exception
geode-core/src/main/java/org/apache/geode/internal/cache/tx/RemotePutAllMessage.java