@Override
  public void txApplyDestroy(Object key, TransactionId txId, TXRmtEvent txEvent,
      boolean inTokenMode, boolean inRI, Operation op, EventID eventId, Object aCallbackArgument,
      List<EntryEventImpl> pendingCallbacks, FilterRoutingInfo filterRoutingInfo,
      ClientProxyMembershipID bridgeContext, boolean isOriginRemote, TXEntryState txEntryState,
      VersionTag versionTag, long tailKey) {
    assert pendingCallbacks != null;
    final boolean isDebugEnabled = logger.isDebugEnabled();

    final LocalRegion owner = _getOwner();

    final boolean isRegionReady = !inTokenMode;
    inTokenMode = isInTokenModeNeeded(owner, inTokenMode);
    final boolean hasRemoteOrigin = !txId.getMemberId().equals(owner.getMyId());
    boolean callbackEventAddedToPending = false;
    IndexManager oqlIndexManager = owner.getIndexManager();
    final boolean locked = owner.lockWhenRegionIsInitializing();
    try {
      RegionEntry re = getEntry(key);
      if (re != null) {
        // Fix for Bug #44431. We do NOT want to update the region and wait
        // later for index INIT as region.clear() can cause inconsistency if
        // happened in parallel as it also does index INIT.
        if (oqlIndexManager != null) {
          oqlIndexManager.waitForIndexInit();
        }
        try {
          synchronized (re) {
            if (!re.isRemoved() || re.isTombstone()) {
              Object oldValue = re.getValueInVM(owner);
              final int oldSize = owner.calculateRegionEntryValueSize(re);
              final boolean wasDestroyedOrRemoved = re.isDestroyedOrRemoved();
              // Create an entry event only if the calling context is
              // a receipt of a TXCommitMessage AND there are callbacks installed
              // for this region
              @Released
              final EntryEventImpl callbackEvent = txCallbackEventFactory
                  .createCallbackEvent(owner, op, key, null, txId,
                      txEvent, eventId, aCallbackArgument, filterRoutingInfo, bridgeContext,
                      txEntryState, versionTag, tailKey);
              try {

                if (owner.isUsedForPartitionedRegionBucket()) {
                  txHandleWANEvent(owner, callbackEvent, txEntryState);
                }
                callbackEvent.setRegionEntry(re);
                callbackEvent.setOldValue(oldValue);
                if (isDebugEnabled) {
                  logger.debug("txApplyDestroy callbackEvent={}", callbackEvent);
                }

                txRemoveOldIndexEntry(Operation.DESTROY, re);
                if (txEvent != null) {
                  txEvent.addDestroy(owner, re, re.getKey(), aCallbackArgument);
                }
                boolean clearOccured = false;
                try {
                  processAndGenerateTXVersionTag(callbackEvent, re, txEntryState);
                  if (inTokenMode) {
                    if (oldValue == Token.TOMBSTONE) {
                      owner.unscheduleTombstone(re);
                    }
                    re.setValue(owner, Token.DESTROYED);
                  } else {
                    if (!re.isTombstone()) {
                      {
                        if (owner.getConcurrencyChecksEnabled()
                            && callbackEvent.getVersionTag() != null) {
                          re.makeTombstone(owner, callbackEvent.getVersionTag());
                        } else {
                          re.removePhase1(owner, false); // fix for bug 43063
                          re.removePhase2();
                          removeEntry(key, re, false);
                        }
                      }
                    } else {
                      owner.rescheduleTombstone(re, re.getVersionStamp().asVersionTag());
                    }
                  }
                  EntryLogger.logTXDestroy(_getOwnerObject(), key);
                  if (!wasDestroyedOrRemoved) {
                    owner.updateSizeOnRemove(key, oldSize);
                  }
                } catch (RegionClearedException rce) {
                  clearOccured = true;
                }
                owner.txApplyDestroyPart2(re, re.getKey(), inTokenMode,
                    clearOccured /* Clear Conflciting with the operation */,
                    wasDestroyedOrRemoved);
                boolean invokeCallbacks = shouldInvokeCallbacks(owner, isRegionReady || inRI);
                if (invokeCallbacks) {
                  switchEventOwnerAndOriginRemote(callbackEvent, hasRemoteOrigin);
                  pendingCallbacks.add(callbackEvent);
                  callbackEventAddedToPending = true;
                }
                if (!clearOccured) {
                  lruEntryDestroy(re);
                }
                if (owner.getConcurrencyChecksEnabled() && txEntryState != null) {
                  txEntryState.setVersionTag(callbackEvent.getVersionTag());
                }
              } finally {
                if (!callbackEventAddedToPending) {
                  releaseEvent(callbackEvent);
                }
              }
            }
          }
        } finally {
          if (oqlIndexManager != null) {
            oqlIndexManager.countDownIndexUpdaters();
          }
        }
      } else if (!isRegionReady || owner.getConcurrencyChecksEnabled()) {
        // treating tokenMode and re == null as same, since we now want to
        // generate versions and Tombstones for destroys
        boolean opCompleted = false;
        RegionEntry newRe = getEntryFactory().createEntry(owner, key, Token.REMOVED_PHASE1);
        if (oqlIndexManager != null) {
          oqlIndexManager.waitForIndexInit();
        }
        EntryEventImpl callbackEvent = null;
        try {
          synchronized (newRe) {
            RegionEntry oldRe = putEntryIfAbsent(key, newRe);
            while (!opCompleted && oldRe != null) {
              synchronized (oldRe) {
                if (oldRe.isRemovedPhase2()) {
                  owner.getCachePerfStats().incRetries();
                  getEntryMap().remove(key, oldRe);
                  oldRe = putEntryIfAbsent(key, newRe);
                } else {
                  try {
                    boolean invokeCallbacks = shouldInvokeCallbacks(owner, isRegionReady || inRI);
                    callbackEvent = txCallbackEventFactory
                        .createCallbackEvent(owner, op, key, null, txId, txEvent,
                            eventId, aCallbackArgument, filterRoutingInfo, bridgeContext,
                            txEntryState,
                            versionTag, tailKey);
                    try {
                      callbackEvent.setRegionEntry(oldRe);
                      callbackEvent.setOldValue(Token.NOT_AVAILABLE);
                      if (isDebugEnabled) {
                        logger.debug("txApplyDestroy token mode callbackEvent={}", callbackEvent);
                      }
                      if (owner.isUsedForPartitionedRegionBucket()) {
                        txHandleWANEvent(owner, callbackEvent, txEntryState);
                      }
                      processAndGenerateTXVersionTag(callbackEvent, oldRe, txEntryState);
                      if (invokeCallbacks) {
                        switchEventOwnerAndOriginRemote(callbackEvent, hasRemoteOrigin);
                        pendingCallbacks.add(callbackEvent);
                        callbackEventAddedToPending = true;
                      }
                      int oldSize = 0;
                      boolean wasTombstone = oldRe.isTombstone();
                      boolean wasDestroyedOrRemoved = oldRe.isDestroyedOrRemoved();
                      {
                        if (!wasTombstone) {
                          oldSize = owner.calculateRegionEntryValueSize(oldRe);
                        }
                      }
                      // TODO: Token.DESTROYED should only be used if "inTokenMode".
                      // Otherwise this should be a TOMBSTONE
                      oldRe.setValue(owner, Token.DESTROYED);
                      EntryLogger.logTXDestroy(_getOwnerObject(), key);
                      if (wasTombstone) {
                        owner.unscheduleTombstone(oldRe);
                      }
                      if (!wasDestroyedOrRemoved) {
                        owner.updateSizeOnRemove(oldRe.getKey(), oldSize);
                      }
                      owner.txApplyDestroyPart2(oldRe, oldRe.getKey(), inTokenMode,
                          false /* Clear Conflicting with the operation */, wasDestroyedOrRemoved);
                      lruEntryDestroy(oldRe);
                    } finally {
                      if (!callbackEventAddedToPending) {
                        releaseEvent(callbackEvent);
                      }
                    }
                  } catch (RegionClearedException rce) {
                    owner.txApplyDestroyPart2(oldRe, oldRe.getKey(), inTokenMode,
                        true /* Clear Conflicting with the operation */, true);
                  }
                  if (owner.getConcurrencyChecksEnabled()
                      && callbackEvent.getVersionTag() != null) {
                    oldRe.makeTombstone(owner, callbackEvent.getVersionTag());
                  } else if (!inTokenMode) {
                    // only remove for NORMAL regions if they do not generate versions see 51781
                    oldRe.removePhase1(owner, false); // fix for bug 43063
                    oldRe.removePhase2();
                    removeEntry(key, oldRe, false);
                  }
                  opCompleted = true;
                }
              }
            }
            if (!opCompleted) {
              opCompleted = true;
              boolean invokeCallbacks = shouldInvokeCallbacks(owner, isRegionReady || inRI);
              callbackEvent = txCallbackEventFactory
                  .createCallbackEvent(owner, op, key, null, txId, txEvent, eventId,
                      aCallbackArgument, filterRoutingInfo, bridgeContext, txEntryState, versionTag,
                      tailKey);
              try {
                callbackEvent.setRegionEntry(newRe);
                callbackEvent.setOldValue(Token.NOT_AVAILABLE);
                if (isDebugEnabled) {
                  logger.debug("txApplyDestroy token mode callbackEvent={}", callbackEvent);
                }
                if (owner.isUsedForPartitionedRegionBucket()) {
                  txHandleWANEvent(owner, callbackEvent, txEntryState);
                }
                processAndGenerateTXVersionTag(callbackEvent, newRe, txEntryState);
                if (invokeCallbacks) {
                  switchEventOwnerAndOriginRemote(callbackEvent, hasRemoteOrigin);
                  pendingCallbacks.add(callbackEvent);
                  callbackEventAddedToPending = true;
                }
                EntryLogger.logTXDestroy(_getOwnerObject(), key);
                if (owner.getConcurrencyChecksEnabled() && callbackEvent.getVersionTag() != null) {
                  newRe.makeTombstone(owner, callbackEvent.getVersionTag());
                } else if (!inTokenMode) {
                  // only remove for NORMAL regions if they do not generate versions see 51781
                  newRe.removePhase1(owner, false); // fix for bug 43063
                  newRe.removePhase2();
                  removeEntry(key, newRe, false);
                } else {
                  newRe.setValue(owner, Token.DESTROYED);
                }
                owner.txApplyDestroyPart2(newRe, newRe.getKey(), inTokenMode,
                    false /* clearConflict */, true);
                // Note no need for LRU work since the entry is destroyed
                // and will be removed when gii completes
              } finally {
                if (!callbackEventAddedToPending) {
                  releaseEvent(callbackEvent);
                }
              }
            }
            if (owner.getConcurrencyChecksEnabled() && txEntryState != null) {
              txEntryState.setVersionTag(callbackEvent.getVersionTag());
            }
          }
        } catch (RegionClearedException e) {
          // TODO
        } finally {
          if (oqlIndexManager != null) {
            oqlIndexManager.countDownIndexUpdaters();
          }
        }
      } else { // re == null
        // Fix bug#43594
        // In cases where bucket region is re-created, it may so happen that
        // the destroy is already applied on the Initial image provider, thus
        // causing region entry to be absent.
        // Notify clients with client events.
        @Released
        EntryEventImpl callbackEvent =
            txCallbackEventFactory
                .createCallbackEvent(owner, op, key, null, txId, txEvent, eventId,
                    aCallbackArgument,
                    filterRoutingInfo, bridgeContext, txEntryState, versionTag, tailKey);
        try {
          if (owner.isUsedForPartitionedRegionBucket()) {
            txHandleWANEvent(owner, callbackEvent, txEntryState);
          }
          switchEventOwnerAndOriginRemote(callbackEvent, hasRemoteOrigin);
          pendingCallbacks.add(callbackEvent);
          callbackEventAddedToPending = true;
        } finally {
          if (!callbackEventAddedToPending) {
            releaseEvent(callbackEvent);
          }
        }
      }
    } catch (DiskAccessException dae) {
      owner.handleDiskAccessException(dae);
      throw dae;
    } finally {
      if (locked) {
        owner.unlockWhenRegionIsInitializing();
      }
    }
  }
Process an incoming version tag for concurrent operation detection. This must be done before
modifying the region entry.

@param regionEntry the entry that is to be modified
@param event the modification to the entry
@throws InvalidDeltaException if the event contains a delta that cannot be applied
@throws ConcurrentCacheModificationException if the event is in conflict with a previously
        applied change
geode-core/src/main/java/org/apache/geode/internal/cache/AbstractRegionMap.java