public void destroyInBucket(final EntryEventImpl event, Object expectedOldValue)
      throws EntryNotFoundException, CacheWriterException {
    // Get the bucket id for the key
    final Integer bucketId = event.getKeyInfo().getBucketId();
    assert bucketId != KeyInfo.UNKNOWN_BUCKET;
    // check in bucket2Node region
    final InternalDistributedMember targetNode = getOrCreateNodeForBucketWrite(bucketId, null);

    if (logger.isDebugEnabled()) {
      logger.debug("destroyInBucket: key={} ({}) in node {} to bucketId={} retry={} ms",
          event.getKey(), event.getKey().hashCode(), targetNode, bucketStringForLogs(bucketId),
          retryTimeout);
    }

    // retry the put remotely until it finds the right node managing the bucket
    RetryTimeKeeper retryTime = null;
    InternalDistributedMember currentTarget = targetNode;
    long timeOut = 0;
    int count = 0;
    for (;;) {
      switch (count) {
        case 0:
          // Note we don't check for DM cancellation in common case.
          // First time, keep going
          break;
        case 1:
          // First failure
          cache.getCancelCriterion().checkCancelInProgress(null);
          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, "destroy 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;
      }
      count++;

      if (currentTarget == null) { // pick target
        checkReadiness();

        if (retryTime == null) {
          retryTime = new RetryTimeKeeper(retryTimeout);
        }
        if (retryTime.overMaximum()) {
          // if (this.getNodeList(bucketId) == null
          // || this.getNodeList(bucketId).size() == 0) {
          // throw new EntryNotFoundException("Entry not found for key "
          // + event.getKey());
          // }
          if (getRegionAdvisor().getBucket(bucketId).getBucketAdvisor()
              .basicGetPrimaryMember() == null) {
            throw new EntryNotFoundException(
                String.format("Entry not found for key %s",
                    event.getKey()));
          }
          TimeoutException e = new TimeoutException(
              String.format("Time out looking for target node for destroy; waited %s ms",
                  retryTime.getRetryTime()));
          if (logger.isDebugEnabled()) {
            logger.debug(e.getMessage(), e);
          }
          checkReadiness();
          throw e;
        }

        currentTarget = getOrCreateNodeForBucketWrite(bucketId, retryTime);

        // No storage found for bucket, early out preventing hot loop, bug 36819
        if (currentTarget == null) {
          checkEntryNotFound(event.getKey());
        }
        continue;
      } // pick target

      final boolean isLocal = (localMaxMemory > 0) && currentTarget.equals(getMyId());
      try {

        DistributedRemoveAllOperation savedOp = event.setRemoveAllOperation(null);
        if (savedOp != null) {
          savedOp.addEntry(event, bucketId);
          return;
        }
        if (isLocal) {
          // doCacheWriteBeforeDestroy(event);
          event.setInvokePRCallbacks(true);
          dataStore.destroyLocally(bucketId, event, expectedOldValue);
        } else {
          if (event.isBridgeEvent()) {
            setNetworkHopType(bucketId, currentTarget);
          }
          destroyRemotely(currentTarget, bucketId, event, expectedOldValue);
        }
        return;

        // NOTREACHED (success)
      } catch (ConcurrentCacheModificationException e) {
        if (logger.isDebugEnabled()) {
          logger.debug("destroyInBucket: caught concurrent cache modification exception", e);
        }
        event.isConcurrencyConflict(true);

        if (logger.isTraceEnabled()) {
          logger.trace(
              "ConcurrentCacheModificationException received for destroyInBucket for bucketId: {}{}{} for event: {} No reattempt is done, returning from here",
              getPRId(), BUCKET_ID_SEPARATOR, bucketId, event);
        }
        return;
      } catch (ForceReattemptException e) {
        e.checkKey(event.getKey());
        // We don't know if the destroy took place or not at this point.
        // Assume that if the next destroy throws EntryDestroyedException, the
        // previous destroy attempt was a success
        checkReadiness();
        InternalDistributedMember lastNode = currentTarget;
        if (retryTime == null) {
          retryTime = new RetryTimeKeeper(retryTimeout);
        }
        currentTarget = getOrCreateNodeForBucketWrite(bucketId, retryTime);
        event.setPossibleDuplicate(true);
        if (lastNode.equals(currentTarget)) {
          if (retryTime.overMaximum()) {
            PRHARedundancyProvider.timedOut(this, null, null, "destroy an entry",
                retryTime.getRetryTime());
          }
          retryTime.waitToRetryNode();
        }
      } catch (PrimaryBucketException notPrimary) {
        if (logger.isDebugEnabled()) {
          logger.debug("destroyInBucket: {} on Node {} not primary",
              notPrimary.getLocalizedMessage(), currentTarget);
        }
        getRegionAdvisor().notPrimary(bucketId, currentTarget);
        if (retryTime == null) {
          retryTime = new RetryTimeKeeper(retryTimeout);
        }
        currentTarget = getOrCreateNodeForBucketWrite(bucketId, retryTime);
      }

      // If we get here, the attempt failed.
      if (count == 1) {
        prStats.incDestroyOpsRetried();
      }
      prStats.incDestroyRetries();

      if (logger.isDebugEnabled()) {
        logger.debug(
            "destroyInBucket: Attempting to resend destroy to node {} after {} failed attempts",
            currentTarget, count);
      }
    } // for
  }
@param expectedOldValue only succeed if current value is equal to expectedOldValue
@throws EntryNotFoundException if entry not found or if expectedOldValue not null and current
        value was not equal to expectedOldValue
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java