@edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "IMSE_DONT_CATCH_IMSE")
  public boolean doLocalRemoveAll(PartitionedRegion r, InternalDistributedMember eventSender,
      boolean cacheWrite)
      throws EntryExistsException, DataLocationException {
    boolean didRemove = false;
    long clientReadTimeOut = PoolFactory.DEFAULT_READ_TIMEOUT;
    if (r.hasServerProxy()) {
      clientReadTimeOut = r.getServerProxy().getPool().getReadTimeout();
      if (logger.isDebugEnabled()) {
        logger.debug("RemoveAllPRMessage: doLocalRemoveAll: clientReadTimeOut is {}",
            clientReadTimeOut);
      }
    }

    DistributedRemoveAllOperation op = null;
    @Released
    EntryEventImpl baseEvent = null;
    BucketRegion bucketRegion = null;
    PartitionedRegionDataStore ds = r.getDataStore();
    InternalDistributedMember myId = r.getDistributionManager().getDistributionManagerId();
    try {

      if (!notificationOnly) {
        // bucketRegion is not null only when !notificationOnly
        bucketRegion = ds.getInitializedBucketForId(null, bucketId);

        versions = new VersionedObjectList(removeAllPRDataSize, true,
            bucketRegion.getAttributes().getConcurrencyChecksEnabled());

        // create a base event and a DPAO for RemoveAllMessage distributed btw redundant buckets
        baseEvent = EntryEventImpl.create(bucketRegion, Operation.REMOVEALL_DESTROY, null, null,
            callbackArg, true, eventSender, !skipCallbacks, true);
        // set baseEventId to the first entry's event id. We need the thread id for DACE
        baseEvent.setEventId(removeAllPRData[0].getEventID());
        if (bridgeContext != null) {
          baseEvent.setContext(bridgeContext);
        }
        baseEvent.setPossibleDuplicate(posDup);
        if (logger.isDebugEnabled()) {
          logger.debug(
              "RemoveAllPRMessage.doLocalRemoveAll: eventSender is {}, baseEvent is {}, msg is {}",
              eventSender, baseEvent, this);
        }
        op = new DistributedRemoveAllOperation(baseEvent, removeAllPRDataSize, false);
      }
      Object[] keys = getKeysToBeLocked();

      if (!notificationOnly) {
        boolean locked = false;
        try {
          if (removeAllPRData.length > 0) {
            if (posDup && bucketRegion.getConcurrencyChecksEnabled()) {
              if (logger.isDebugEnabled()) {
                logger.debug("attempting to locate version tags for retried event");
              }
              // bug #48205 - versions may have already been generated for a posdup event
              // so try to recover them before wiping out the eventTracker's record
              // of the previous attempt
              for (int i = 0; i < removeAllPRDataSize; i++) {
                if (removeAllPRData[i].versionTag == null) {
                  removeAllPRData[i].versionTag =
                      bucketRegion.findVersionTagForClientBulkOp(removeAllPRData[i].getEventID());
                  if (removeAllPRData[i].versionTag != null) {
                    removeAllPRData[i].versionTag.replaceNullIDs(bucketRegion.getVersionMember());
                  }
                }
              }
            }
            EventID eventID = removeAllPRData[0].getEventID();
            ThreadIdentifier membershipID =
                new ThreadIdentifier(eventID.getMembershipID(), eventID.getThreadID());
            bucketRegion.recordBulkOpStart(membershipID, eventID);
          }
          locked = bucketRegion.waitUntilLocked(keys);
          boolean lockedForPrimary = false;
          final ArrayList<Object> succeeded = new ArrayList<>();
          PutAllPartialResult partialKeys = new PutAllPartialResult(removeAllPRDataSize);
          Object key = keys[0];
          try {
            bucketRegion.doLockForPrimary(false);
            lockedForPrimary = true;

            /*
             * The real work to be synchronized, it will take long time. We don't worry about
             * another thread to send any msg which has the same key in this request, because these
             * request will be blocked by foundKey
             */
            for (int i = 0; i < removeAllPRDataSize; i++) {
              @Released
              EntryEventImpl ev = getEventFromEntry(r, myId, eventSender, i, removeAllPRData,
                  notificationOnly, bridgeContext, posDup, skipCallbacks);
              try {
                key = ev.getKey();

                ev.setRemoveAllOperation(op);

                // ev will be added into the op in removeLocally()
                // real operation will be modified into ev in removeLocally()
                // then in basicPutPart3(), the ev is added into op
                try {
                  r.getDataView().destroyOnRemote(ev, cacheWrite, null);
                  didRemove = true;
                  if (logger.isDebugEnabled()) {
                    logger.debug(
                        "RemoveAllPRMessage.doLocalRemoveAll:removeLocally success for " + ev);
                  }
                } catch (EntryNotFoundException ignore) {
                  didRemove = true;
                  if (ev.isPossibleDuplicate() && ev.hasValidVersionTag()) {
                    op.addEntry(ev);
                    if (logger.isDebugEnabled()) {
                      logger.debug(
                          "RemoveAllPRMessage.doLocalRemoveAll:notify client and gateway for not-found-entry:"
                              + ev);
                    }
                  }
                  if (ev.getVersionTag() == null) {
                    if (logger.isDebugEnabled()) {
                      logger.debug(
                          "doLocalRemoveAll:RemoveAll encoutered EntryNotFoundException: event={}",
                          ev);
                    }
                  }
                } catch (ConcurrentCacheModificationException e) {
                  didRemove = true;
                  if (logger.isDebugEnabled()) {
                    logger.debug(
                        "RemoveAllPRMessage.doLocalRemoveAll:removeLocally encountered concurrent cache modification for "
                            + ev);
                  }
                }
                removeAllPRData[i].setTailKey(ev.getTailKey());
                if (!didRemove) { // make sure the region hasn't gone away
                  r.checkReadiness();
                  ForceReattemptException fre = new ForceReattemptException(
                      "unable to perform remove in RemoveAllPR, but operation should not fail");
                  fre.setHash(ev.getKey().hashCode());
                  throw fre;
                } else {
                  succeeded.add(removeAllPRData[i].getKey());
                  versions.addKeyAndVersion(removeAllPRData[i].getKey(), ev.getVersionTag());
                }
              } finally {
                ev.release();
              }
            } // for

          } catch (IllegalMonitorStateException ex) {
            throw new ForceReattemptException("unable to get lock for primary, retrying... ");
          } catch (CacheWriterException cwe) {
            // encounter cacheWriter exception
            partialKeys.saveFailedKey(key, cwe);
          } finally {
            doPostRemoveAll(r, op, bucketRegion, lockedForPrimary);
          }
          if (partialKeys.hasFailure()) {
            partialKeys.addKeysAndVersions(versions);
            if (logger.isDebugEnabled()) {
              logger.debug(
                  "RemoveAllPRMessage: partial keys applied, map to bucket {}'s keys:{}. Applied {}",
                  bucketId, Arrays.toString(keys), succeeded);
            }
            throw new PutAllPartialResultException(partialKeys);
          }
        } catch (RegionDestroyedException e) {
          ds.checkRegionDestroyedOnBucket(bucketRegion, true, e);
        } finally {
          if (locked) {
            bucketRegion.removeAndNotifyKeys(keys);
          }
        }
      } else {
        for (int i = 0; i < removeAllPRDataSize; i++) {
          EntryEventImpl ev = getEventFromEntry(r, myId, eventSender, i, removeAllPRData,
              notificationOnly, bridgeContext, posDup, skipCallbacks);
          try {
            ev.setOriginRemote(true);
            if (callbackArg != null) {
              ev.setCallbackArgument(callbackArg);
            }
            r.invokeDestroyCallbacks(EnumListenerEvent.AFTER_DESTROY, ev, r.isInitialized(), true);
          } finally {
            ev.release();
          }
        }
      }
    } finally {
      if (baseEvent != null) {
        baseEvent.release();
      }
      if (op != null) {
        op.freeOffHeapResources();
      }
    }

    return true;
  }
This method is called by both operateOnPartitionedRegion() when processing a remote msg or by
sendMsgByBucket() when processing a msg targeted to local Jvm. PartitionedRegion Note: It is
very important that this message does NOT cause any deadlocks as the sender will wait
indefinitely for the acknowledgment

@param r partitioned region
@param eventSender the endpoint server who received request from client
@param cacheWrite if true invoke cacheWriter before desrtoy
@return If succeeds, return true, otherwise, throw exception
geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/RemoveAllPRMessage.java