@Override
  public void basicDestroy(EntryEventImpl event, boolean cacheWrite, Object expectedOldValue)
      throws EntryNotFoundException, CacheWriterException, TimeoutException {
    // disallow local destruction for mirrored keysvalues regions
    boolean hasSeen = hasSeenEvent(event);
    checkIfReplicatedAndLocalDestroy(event);

    try {
      boolean invokeWriter = cacheWrite;
      if (requiresOneHopForMissingEntry(event)) {
        // bug #45704: see if a one-hop must be done for this operation
        RegionEntry re = getRegionEntry(event.getKey());
        if (re == null /* || re.isTombstone() */ || !generateVersionTag) {
          if (getServerProxy() == null) {
            // only assert for non-client regions.
            Assert.assertTrue(!getDataPolicy().withReplication() || !generateVersionTag);
          }
          if (!event.isBulkOpInProgress() || getDataPolicy().withStorage()) {
            // removeAll will send a single one-hop for empty regions. for other missing entries
            // we need to get a valid version number before modifying the local cache
            // TODO: deltaGII: verify that delegating to a peer when this region is also a client is
            // acceptable
            boolean didDistribute =
                RemoteDestroyMessage.distribute(event, expectedOldValue, !generateVersionTag);

            if (!generateVersionTag && !didDistribute) {
              throw new PersistentReplicatesOfflineException();
            }

            if (didDistribute) {
              if (logger.isTraceEnabled()) {
                logger.trace("Event after remoteDestroy operation: {}", event);
              }
              invokeWriter = false; // remote cache invoked the writer
              if (event.getVersionTag() == null) {
                // if the event wasn't applied by the one-hop replicate it will not have a version
                // tag
                // and so should not be applied to this cache
                return;
              }
            }
          }
        }
      }

      super.basicDestroy(event, invokeWriter, expectedOldValue);

      // if this is a destroy coming in from remote source, free up lock resources
      // if this is a local origin destroy, this will happen after lock is
      // released
      if (scope.isGlobal() && event.isOriginRemote()) {
        try {
          getLockService().freeResources(event.getKey());
        } catch (LockServiceDestroyedException ignore) {
        }
      }

    } finally {
      if (hasSeen) {
        if (event.isBulkOpInProgress() && !event.isOriginRemote()) {
          event.getRemoveAllOperation().addEntry(event, true);
        }
        if (!getConcurrencyChecksEnabled() || event.hasValidVersionTag()) {
          // before distribute: DR.destroy, hasSeenEvent. no to notifyGateway
          distributeDestroy(event, expectedOldValue);
          event.invokeCallbacks(this, true, false);
        }
      }
    }
  }
remove any partial entries received in a failed GII
geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRegion.java