/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.internal.cache.map;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.same;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;

import java.util.Map;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.mockito.InOrder;

import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.query.internal.index.IndexManager;
import org.apache.geode.internal.cache.CachePerfStats;
import org.apache.geode.internal.cache.DistributedRegion;
import org.apache.geode.internal.cache.EntryEventImpl;
import org.apache.geode.internal.cache.InternalRegion;
import org.apache.geode.internal.cache.RegionClearedException;
import org.apache.geode.internal.cache.RegionEntry;
import org.apache.geode.internal.cache.RegionEntryFactory;
import org.apache.geode.internal.cache.Token;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.versions.ConcurrentCacheModificationException;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
import org.apache.geode.internal.cache.versions.VersionStamp;
import org.apache.geode.internal.cache.versions.VersionTag;

public class RegionMapDestroyTest {

  private static final Object KEY = "key";

  @SuppressWarnings("rawtypes")
  private Map entryMap;
  private FocusedRegionMap regionMap;
  private boolean withConcurrencyChecks;
  private RegionEntryFactory factory;
  private RegionEntry newRegionEntry;
  private RegionEntry existingRegionEntry;
  private InternalRegion owner;

  private EntryEventImpl event;
  private Object expectedOldValue;

  private boolean inTokenMode;
  private boolean duringRI;
  private boolean cacheWrite;
  private boolean isEviction;
  private boolean removeRecoveredEntry;
  private boolean fromRILocalDestroy;

  private RegionMapDestroy regionMapDestroy;

  private Throwable doDestroyThrowable;

  private boolean doDestroyResult;

  private Runnable testHook;

  private IndexManager indexManager;

  @SuppressWarnings("unchecked")
  @Before
  public void setUp() throws Exception {
    factory = mock(RegionEntryFactory.class);
    entryMap = mock(Map.class);
    newRegionEntry = mock(RegionEntry.class);
    givenEntryDestroyReturnsTrue(newRegionEntry);
    when(newRegionEntry.isDestroyedOrRemoved()).thenReturn(true);
    existingRegionEntry = mock(RegionEntry.class);
    givenEntryDestroyReturnsTrue(existingRegionEntry);

    owner = mock(InternalRegion.class);
    when(owner.getCachePerfStats()).thenReturn(mock(CachePerfStats.class));
    when(owner.getDataPolicy()).thenReturn(DataPolicy.REPLICATE);
    when(owner.getConcurrencyChecksEnabled()).thenReturn(withConcurrencyChecks);
    // Instead of mocking checkEntryNotFound to throw an exception,
    // this test now just verifies that checkEntryNotFound was called.
    // Having the mock throw the exception confuses the code coverage tools.

    regionMap = mock(FocusedRegionMap.class);
    when(regionMap.getEntryFactory()).thenReturn(factory);
    when(factory.createEntry(any(), any(), any())).thenReturn(newRegionEntry);
    when(regionMap.getEntryMap()).thenReturn(entryMap);

    event = mock(EntryEventImpl.class);
    when(event.getRegion()).thenReturn(owner);
    when(event.getOperation()).thenReturn(Operation.DESTROY);
    when(event.getKey()).thenReturn(KEY);
    when(event.isGenerateCallbacks()).thenReturn(true);

    inTokenMode = false;
    duringRI = false;
    cacheWrite = false;
    isEviction = false;
    expectedOldValue = null;
    removeRecoveredEntry = false;

    regionMapDestroy = new RegionMapDestroy(owner, regionMap, mock(CacheModificationLock.class));
  }

  @After
  public void tearDown() {
    RegionMapDestroy.testHookRunnableForConcurrentOperation = null;
  }

  @Test
  public void destroyWithDuplicateVersionInvokesListener() {
    givenConcurrencyChecks(true);
    givenExistingTombstone();
    givenExistingEntryWithVersionTag(mock(VersionTag.class));
    givenEventWithClientOrigin();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(event, times(1)).setIsRedestroyedEntry(true);
    verifyPart3();
  }

  @Test
  public void destroyWithEmptyRegionThrowsException() {
    givenConcurrencyChecks(false);

    doDestroy();

    verify(owner, times(1)).checkEntryNotFound(any());
  }

  @Test
  public void destroyWithEmptyRegionInTokenModeAddsAToken() throws Exception {
    givenConcurrencyChecks(false);
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedTrue();
    verifyEntryAddedToMap(newRegionEntry);
    verifyEntryDestroyed(newRegionEntry, true);
    verifyInvokedDestroyMethodsOnRegion(false);
  }

  @Test
  public void destroyWithEmptyRegionInTokenModeNeverCallsUpdateSizeOnRemove() {
    givenConcurrencyChecks(false);
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(owner, never()).updateSizeOnRemove(any(), anyInt());
  }

  @Test
  public void destroyInvokesTestHook() {
    givenConcurrencyChecks(false);
    givenInTokenMode();
    givenTestHook();

    doDestroy();

    verifyTestHookRun();
  }

  @Test
  public void destroyWithEmptyRegionInTokenModeWithRegionClearedExceptionDoesDestroy()
      throws Exception {
    givenConcurrencyChecks(false);
    givenDestroyThrowsRegionClearedException();
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedTrue();
    verifyInvokedDestroyMethodsOnRegion(true);
  }

  @Test
  public void evictDestroyWithEmptyRegionInTokenModeDoesNothing() {
    givenEviction();
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedFalse();
    verifyEntryRemoved(newRegionEntry);
    verifyNoDestroyInvocationsOnRegion();
  }

  @Test
  public void evictDestroyWithExistingTombstoneInTokenModeDestroyExistingEntry() throws Exception {
    givenConcurrencyChecks(true);
    givenEviction();
    givenExistingTombstone();
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedTrue();
    verifyEntryDestroyed(existingRegionEntry, false);
    verifyInvokedDestroyMethodsOnRegion(false);
  }

  @Test
  public void evictDestroyWithExistingTombstoneInTokenModeNeverCallsUpdateSizeOnRemove() {
    givenConcurrencyChecks(true);
    givenEviction();
    givenExistingTombstone();
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(owner, never()).updateSizeOnRemove(any(), anyInt());
  }

  @Test
  public void evictDestroyWithExistingTombstoneInUseByTransactionInTokenModeDoesNothing()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenEviction();
    givenExistingEntryWithValue(Token.TOMBSTONE);
    givenEntryIsInUseByTransaction();
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedFalse();
    verifyNoDestroyInvocationsOnEntry(existingRegionEntry);
    verifyNoDestroyInvocationsOnRegion();
  }

  @Test
  public void evictDestroyWithConcurrentChangeFromNullToInUseByTransactionInTokenModeDoesNothing()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenEviction();
    givenExistingEntryWithValue(null);
    givenEntryIsInUseByTransaction();
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedFalse();
    verifyNoDestroyInvocationsOnEntry(existingRegionEntry);
    verifyNoDestroyInvocationsOnRegion();
  }

  @Test
  public void evictDestroyWithInUseByTransactionInTokenModeDoesNothing()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenEviction();
    givenExistingEntry();
    givenEntryIsInUseByTransaction();
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedFalse();
    verifyNoDestroyInvocationsOnEntry(existingRegionEntry);
    verifyNoDestroyInvocationsOnRegion();
  }

  @Test
  public void destroyWithConcurrentChangeFromNullToValidRetriesAndDoesDestroy()
      throws Exception {
    givenConcurrencyChecks(true);
    givenMissThenExistingEntry();

    doDestroy();

    verifyDestroyReturnedTrue();
    verifyEntryDestroyed(existingRegionEntry, false);
    verifyInvokedDestroyMethodsOnRegion(false);
  }

  @Test
  public void destroyWithConcurrentChangeFromNullToValidRetriesAndThrowsConcurrentCacheModificationException()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenMissThenExistingEntry();
    givenEntryDestroyThrows(existingRegionEntry, ConcurrentCacheModificationException.class);

    doDestroyExpectingThrowable();

    verifyThrowableInstanceOf(ConcurrentCacheModificationException.class);
    verifyNoDestroyInvocationsOnRegion();
  }

  @Test
  public void destroyWithConcurrentChangeFromNullToValidRetriesCallsDestroyWhichReturnsFalseCausingDestroyToNotHappen()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenMissThenExistingEntry();
    givenExistingEntryWithNoVersionStamp();
    givenEntryDestroyReturnsFalse(existingRegionEntry);

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(existingRegionEntry, times(1)).removePhase2();
    verifyNoDestroyInvocationsOnRegion();
  }

  @Test
  public void destroyWithInTokenModeAndTombstoneCallsDestroyWhichReturnsFalseCausingDestroyToNotHappen()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenExistingEntryWithValue(Token.TOMBSTONE);
    givenMissThenExistingEntry();
    givenInTokenMode();
    givenEntryDestroyReturnsFalse(existingRegionEntry);

    doDestroy();

    // TODO since destroy returns false it seems like doDestroy should return false
    verifyDestroyReturnedTrue();
    verifyNoDestroyInvocationsOnRegion();
  }

  @Test
  public void destroyWithInTokenModeAndTombstoneCallsDestroyWhichThrowsRegionClearedStillDoesDestroy()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenMissThenExistingTombstone();
    givenInTokenMode();
    givenEntryDestroyThrows(existingRegionEntry, RegionClearedException.class);

    doDestroy();

    verifyDestroyReturnedTrue();
    verifyInvokedDestroyMethodsOnRegion(true);
  }

  @Test
  public void destroyWithInTokenModeCallsDestroyWhichReturnsFalseCausingDestroyToNotHappen()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenExistingEntry();
    givenEntryDestroyReturnsFalse(existingRegionEntry);
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedFalse();
    verify(existingRegionEntry, never()).removePhase2();
    verifyNoDestroyInvocationsOnRegion();
  }

  @Test
  public void destroyExistingEntryWithVersionStampCallsDestroyWhichReturnsFalseCausingDestroyToNotHappenAndDoesNotCallRemovePhase2()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenExistingEntry();
    givenEntryDestroyReturnsFalse(existingRegionEntry);
    givenOriginIsRemote();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(owner, never()).rescheduleTombstone(any(), any());
    verify(existingRegionEntry, never()).removePhase2();
    verifyNoDestroyInvocationsOnRegion();
  }

  @Test
  public void destroyTombstoneWithRemoveRecoveredEntryAndVersionStampCallsRescheduleTombstone()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenEviction();
    givenExistingEntryWithValue(Token.TOMBSTONE);
    givenEntryDestroyReturnsFalse(existingRegionEntry);
    givenOriginIsRemote();
    givenRemoveRecoveredEntry();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(owner, times(1)).rescheduleTombstone(same(existingRegionEntry), any());
    verify(existingRegionEntry, never()).removePhase2();
    verifyNoDestroyInvocationsOnRegion();
  }

  @Test
  public void destroyTombstoneWithLocalOriginAndRemoveRecoveredEntryAndVersionStampDoesNotCallRescheduleTombstone()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenEviction();
    givenExistingEntryWithValue(Token.TOMBSTONE);
    givenEntryDestroyReturnsFalse(existingRegionEntry);
    givenRemoveRecoveredEntry();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(owner, never()).rescheduleTombstone(any(), any());
    verify(existingRegionEntry, never()).removePhase2();
    verifyNoDestroyInvocationsOnRegion();
  }

  @Test
  public void destroyWithConcurrentChangeFromNullToValidRetriesAndCallsUpdateSizeOnRemove()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenMissThenExistingEntry();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(owner, times(1)).updateSizeOnRemove(any(), anyInt());
  }

  @Test
  public void destroyInTokenModeWithConcurrentChangeFromNullToRemovePhase2RetriesAndDoesDestroy()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenExistingEntry();
    givenInTokenMode();
    givenRemovePhase2Retry();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(entryMap).remove(eq(KEY), eq(existingRegionEntry));
    verify(regionMap, times(2)).putEntryIfAbsent(eq(KEY), any());
    verifyNoDestroyInvocationsOnEntry(existingRegionEntry);
    verifyInvokedDestroyMethodsOnRegion(false);
  }

  @Test
  public void destroyInTokenModeWithConcurrentChangeFromNullToRemovePhase2RetriesAndNeverCallsUpdateSizeOnRemove()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenExistingEntry();
    givenInTokenMode();
    givenRemovePhase2Retry();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(owner, never()).updateSizeOnRemove(any(), anyInt());
  }

  @Test
  public void destroyOfExistingEntryInTokenModeAddsAToken() throws Exception {
    givenConcurrencyChecks(false);
    givenExistingEntry();
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedTrue();
    verifyEntryDestroyed(existingRegionEntry, false);
    verifyInvokedDestroyMethodsOnRegion(false);
  }

  @Test
  public void destroyOfExistingEntryInTokenModeInhibitsCacheListenerNotification() {
    givenConcurrencyChecks(false);
    givenExistingEntry();
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(event, times(1)).inhibitCacheListenerNotification(true);
  }

  @Test
  public void destroyOfExistingEntryInTokenModeDuringRegisterInterestDoesNotInhibitCacheListenerNotification() {
    givenConcurrencyChecks(false);
    givenExistingEntry();
    givenInTokenMode();
    givenDuringRegisterInterest();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(event, never()).inhibitCacheListenerNotification(true);
  }

  @Test
  public void destroyOfExistingEntryDoesNotInhibitCacheListenerNotification() {
    givenConcurrencyChecks(false);
    givenExistingEntry();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(event, never()).inhibitCacheListenerNotification(true);
  }

  @Test
  public void destroyOfExistingEntryCallsIndexManager() {
    givenConcurrencyChecks(false);
    givenExistingEntry();
    givenIndexManager();

    doDestroy();

    verifyDestroyReturnedTrue();
    verifyIndexManagerOrder();
  }

  @Test
  public void destroyOfExistingEntryInTokenModeCallsUpdateSizeOnRemove() {
    givenConcurrencyChecks(false);
    givenExistingEntry();
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(owner, times(1)).updateSizeOnRemove(any(), anyInt());
  }

  @Test
  public void destroyOfExistingTombstoneInTokenModeWithConcurrencyChecksDoesNothing()
      throws Exception {
    givenConcurrencyChecks(true);
    givenExistingTombstoneAndVersionTag();
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedTrue();
    // why not DESTROY token? since it was already destroyed why do we do the parts?
    verifyEntryDestroyed(existingRegionEntry, false);
    verify(regionMap, never()).removeEntry(eq(KEY), same(existingRegionEntry), anyBoolean());
    verifyInvokedDestroyMethodsOnRegion(false);
  }

  @Test
  public void destroyOfExistingFromPutIfAbsentWithRemoteOriginCallsBasicDestroyBeforeRemoval()
      throws Exception {
    givenConcurrencyChecks(true);
    givenMissThenExistingEntry();
    givenOriginIsRemote();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(owner, times(1)).basicDestroyBeforeRemoval(existingRegionEntry, event);
  }

  @Test
  public void destroyOfExistingFromPutIfAbsentWithTokenModeAndLocalOriginDoesNotCallBasicDestroyBeforeRemoval()
      throws Exception {
    givenConcurrencyChecks(true);
    givenMissThenExistingEntry();
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedTrue();
    // TODO: this seems like a bug in the product. See the comment in:
    // RegionMapDestroy.destroyExistingFromPutIfAbsent(RegionEntry)
    verify(owner, never()).basicDestroyBeforeRemoval(existingRegionEntry, event);
  }

  @Test
  public void destroyOfExistingTombstoneInTokenModeWithConcurrencyChecksNeverCallsUpdateSizeOnRemove() {
    givenConcurrencyChecks(true);
    givenExistingTombstoneAndVersionTag();
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(owner, never()).updateSizeOnRemove(any(), anyInt());
  }

  @Test
  public void destroyOfExistingTombstoneWillThrowConcurrentCacheModificationException()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenExistingTombstoneAndVersionTag();
    givenInTokenMode();
    givenEntryDestroyThrows(existingRegionEntry, ConcurrentCacheModificationException.class);

    doDestroyExpectingThrowable();

    verifyThrowableInstanceOf(ConcurrentCacheModificationException.class);
    verify(owner, never()).notifyTimestampsToGateways(event);
  }

  @Test
  public void destroyOfExistingTombstoneWithTimeStampUpdatedWillCallNotifyTimestampsToGateways()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenExistingTombstoneAndVersionTag();
    givenInTokenMode();
    givenEntryDestroyThrows(existingRegionEntry, ConcurrentCacheModificationException.class);
    givenEventTimeStampUpdated();

    doDestroyExpectingThrowable();

    verifyThrowableInstanceOf(ConcurrentCacheModificationException.class);
    verify(owner, times(1)).notifyTimestampsToGateways(event);
  }

  @Test
  public void destroyOfExistingTombstoneWithConcurrencyChecksThrowsEntryNotFound() {
    givenConcurrencyChecks(true);
    givenExistingTombstoneAndVersionTag();

    doDestroy();

    verify(owner, times(1)).checkEntryNotFound(any());
  }

  @Test
  public void destroyOfExistingTombstoneThatThrowsConcurrentCacheModificationExceptionNeverCallsNotify() {
    givenConcurrencyChecks(true);
    givenExistingTombstone();
    givenVersionStampThatDetectsConflict();
    givenEventWithVersionTag();

    doDestroyExpectingThrowable();

    verifyThrowableInstanceOf(ConcurrentCacheModificationException.class);
    verify(owner, never()).notifyTimestampsToGateways(any());
  }

  @Test
  public void destroyOfTombstoneThatBecomesNonTombstoneRetriesAndDoesDestroy() throws Exception {
    givenConcurrencyChecks(true);
    givenExistingTombstone();
    givenEventWithVersionTag();
    givenTombstoneThenAlive();
    givenNotDestroyedOrRemoved();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(regionMap, times(2)).getEntry(any());
    verify(regionMap, times(1)).processVersionTag(existingRegionEntry, event);
    verify(regionMap, times(1)).lruEntryDestroy(existingRegionEntry);
    verifyNoMoreInteractions(regionMap);
    verifyEntryDestroyed(existingRegionEntry, false);
    verifyInvokedDestroyMethodsOnRegion(false);
  }

  @Test
  public void destroyOfExistingTombstoneThatThrowsConcurrentCacheModificationExceptionWithTimeStampUpdatedCallsNotify()
      throws Exception {
    givenConcurrencyChecks(true);
    givenExistingTombstone();
    givenVersionStampThatDetectsConflict();
    givenEventWithVersionTag();
    givenEventTimeStampUpdated();

    doDestroyExpectingThrowable();

    verifyThrowableInstanceOf(ConcurrentCacheModificationException.class);
    verify(owner, times(1)).notifyTimestampsToGateways(eq(event));
  }

  @Test
  public void destroyWithConcurrentChangeFromNullToValidRetriesAndCallsNotifyTimestampsToGateways()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenMissThenExistingEntry();
    givenVersionStampThatDetectsConflict();
    givenEventWithVersionTag();
    givenEventTimeStampUpdated();
    givenEntryDestroyThrows(existingRegionEntry, ConcurrentCacheModificationException.class);

    doDestroyExpectingThrowable();

    verifyThrowableInstanceOf(ConcurrentCacheModificationException.class);
    verifyNoDestroyInvocationsOnRegion();
    verify(owner, times(1)).notifyTimestampsToGateways(eq(event));
  }

  @Test
  public void destroyOfExistingTombstoneWithConcurrencyChecksAndNoTagThrowsEntryNotFound() {
    givenConcurrencyChecks(true);
    givenExistingTombstone();

    doDestroy();

    verify(owner, times(1)).checkEntryNotFound(any());
  }

  @Test
  public void destroyOfExistingTombstoneDoesDestroyAndReschedulesTombstone()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenExistingTombstone();
    givenEventWithVersionTag();
    givenOriginIsRemote();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(owner, times(1)).checkEntryNotFound(any());
    verify(owner, times(1)).recordEvent(event);
    verify(owner, times(1)).rescheduleTombstone(same(existingRegionEntry), any());
    verify(existingRegionEntry, times(1)).setValue(owner, Token.TOMBSTONE);
    verifyPart2(true);
    verifyNoPart3();
  }

  @Test
  public void evictDestroyOfExistingTombstoneWithConcurrencyChecksReturnsFalse() {
    givenConcurrencyChecks(true);
    givenExistingTombstoneAndVersionTag();
    givenEviction();

    doDestroy();

    verifyDestroyReturnedFalse();
  }

  @Test
  public void destroyOfExistingTombstoneWithConcurrencyChecksAndRemoveRecoveredEntryDoesRemove()
      throws Exception {
    givenConcurrencyChecks(true);
    givenExistingTombstoneAndVersionTag();
    givenTombstoneThenAlive();
    givenRemoveRecoveredEntry();

    doDestroy();

    verifyDestroyReturnedTrue();
    verifyMapDoesNotContainKey();
    verifyInvokedDestroyMethodsOnRegion(false);
  }

  @Test
  public void destroyOfExistingEntryThatBecomesTombstoneAfterInitialCheckCallsProcessVersionTagAndMakeTombstoneButDoesNotDoDestroy()
      throws Exception {
    givenConcurrencyChecks(true);
    givenExistingTombstoneAndVersionTag();
    givenAliveThenTombstone();

    doDestroy();

    verifyDestroyReturnedFalse();
    verify(regionMap, times(1)).getEntry(any());
    verify(regionMap, times(1)).processVersionTag(existingRegionEntry, event);
    verifyNoMoreInteractions(regionMap);
    verify(existingRegionEntry, times(1)).makeTombstone(any(), any());
    verifyNoDestroyInvocationsOnRegion();
  }

  @Test
  public void destroyOfExistingTombstoneWithConcurrencyChecksAndFromRILocalDestroyDoesRemove()
      throws Exception {
    givenConcurrencyChecks(true);
    givenFromRILocalDestroy();
    givenExistingTombstoneAndVersionTag();
    givenTombstoneThenAlive();

    doDestroy();

    verifyDestroyReturnedTrue();
    verifyMapDoesNotContainKey();
    verifyInvokedDestroyMethodsOnRegion(false);
  }

  @Test
  public void destroyOfExistingTombstoneWithConcurrencyChecksAndRemoveRecoveredEntryNeverCallsUpdateSizeOnRemove() {
    givenConcurrencyChecks(true);
    givenExistingTombstoneAndVersionTag();
    givenRemoveRecoveredEntry();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(owner, never()).updateSizeOnRemove(any(), anyInt());
  }

  @Test
  public void destroyOfExistingRemovePhase2WithConcurrencyChecksAndRemoveRecoveredEntryDoesRetryAndThrowsEntryNotFound() {
    givenConcurrencyChecks(true);
    givenExistingEntryWithTokenAndVersionTag(Token.REMOVED_PHASE2);
    givenRemoveRecoveredEntry();

    doDestroy();

    verify(owner, times(1)).checkEntryNotFound(any());
  }

  @Test
  public void destroyOfExistingRemovePhase2WithConcurrencyChecksDoesRetryAndThrowsEntryNotFound() {
    givenConcurrencyChecks(true);
    givenExistingEntryWithTokenAndVersionTag(Token.REMOVED_PHASE2);

    doDestroy();

    verify(owner, times(1)).checkEntryNotFound(any());
  }

  @Test
  public void destroyOfExistingRemovePhase2WithConcurrencyChecksAndExpectedValueDoesRetryAndReturnsFalse() {
    givenConcurrencyChecks(true);
    givenExistingEntryWithTokenAndVersionTag(Token.REMOVED_PHASE2);
    givenExpectedOldValue();

    doDestroy();

    verifyDestroyReturnedFalse();
  }

  @Test
  public void destroyOfExistingRemovePhase2WithConcurrencyChecksAndInTokenModeDoesRetryAndReturnsFalse() {
    givenConcurrencyChecks(true);
    givenExistingEntryWithTokenAndVersionTag(Token.REMOVED_PHASE2);
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedFalse();
  }

  @Test
  public void destroyOfExistingRemovePhase2WithConcurrencyChecksAndEvictionDoesRetryAndReturnsFalse() {
    givenConcurrencyChecks(true);
    givenExistingEntryWithTokenAndVersionTag(Token.REMOVED_PHASE2);
    givenEviction();

    doDestroy();

    verifyDestroyReturnedFalse();
  }

  @Test
  public void destroyOfExistingRemovePhase2WithoutConcurrencyChecksDoesRetryAndThrowsEntryNotFound() {
    givenConcurrencyChecks(false);
    givenExistingEntryWithTokenAndVersionTag(Token.REMOVED_PHASE2);

    doDestroy();

    verify(owner, times(1)).checkEntryNotFound(any());
  }

  @Test
  public void destroyOfExistingRemovePhase2WithConcurrencyChecksAndOriginRemoteDoesRetryAndDoesRemove()
      throws Exception {
    givenConcurrencyChecks(true);
    givenExistingEntryWithTokenAndVersionTag(Token.REMOVED_PHASE2);
    givenOriginIsRemote();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(entryMap, times(1)).remove(KEY, existingRegionEntry);
    verifyEntryAddedToMap(newRegionEntry);
    verifyEntryDestroyed(newRegionEntry, true);
  }

  @Test
  public void destroyOfExistingRemovePhase2WithConcurrencyChecksAndClientOriginDoesRetryAndDoesRemove()
      throws Exception {
    givenConcurrencyChecks(true);
    givenExistingEntryWithTokenAndVersionTag(Token.REMOVED_PHASE2);
    givenEventWithClientOrigin();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(entryMap, times(1)).remove(KEY, existingRegionEntry);
    verifyEntryAddedToMap(newRegionEntry);
    verifyEntryDestroyed(newRegionEntry, true);
  }

  @Test
  public void destroyOfExistingEntryRemovesEntryFromMapAndDoesNotifications() throws Exception {
    givenConcurrencyChecks(false);
    givenExistingEntry();

    doDestroy();

    verifyDestroyReturnedTrue();
    verifyEntryDestroyed(existingRegionEntry, false);
    verify(regionMap, times(1)).removeEntry(KEY, existingRegionEntry, true);
    verifyInvokedDestroyMethodsOnRegion(false);
  }

  @Test
  public void destroyOfExistingEntryWithConflictDoesPart3() {
    givenConcurrencyChecks(false);
    givenExistingEntry();
    givenEventWithConflict();

    doDestroy();

    verifyDestroyReturnedTrue();
    verifyPart3();
  }

  @Test
  public void destroyOfExistingEntryWithConflictAndWANSkipsPart3() {
    givenConcurrencyChecks(false);
    givenExistingEntry();
    givenEventWithConflict();
    givenEventWithGatewayTag();

    doDestroy();

    verifyDestroyReturnedTrue();
    verifyNoPart3();
  }

  @Test
  public void destroyOfExistingEntryWithRegionClearedExceptionDoesDestroyAndPart2AndPart3()
      throws RegionClearedException {
    givenConcurrencyChecks(false);
    givenMissThenExistingEntry();
    givenEventWithVersionTag();
    givenEntryDestroyThrows(existingRegionEntry, RegionClearedException.class);

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(event, never()).inhibitCacheListenerNotification(true);
    verifyInvokedDestroyMethodsOnRegion(true);
  }

  @Test
  public void destroyOfExistingEntryWithRegionClearedExceptionInTokenModeCallsInhibitCacheListenerNotification()
      throws RegionClearedException {
    givenConcurrencyChecks(false);
    givenExistingEntry();
    givenEventWithVersionTag();
    givenEntryDestroyThrows(existingRegionEntry, RegionClearedException.class);
    givenInTokenMode();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(event, times(1)).inhibitCacheListenerNotification(true);
    verifyInvokedDestroyMethodsOnRegion(true);
  }

  @Test
  public void expireDestroyOfExistingEntry() {
    givenConcurrencyChecks(true);
    givenExistingEntry();
    givenExpireDestroy();

    doDestroy();

    verifyDestroyReturnedTrue();
  }

  @Test
  public void expireDestroyOfExistingEntryWithOriginRemote() {
    givenConcurrencyChecks(true);
    givenExistingEntry();
    givenOriginIsRemote();
    givenExpireDestroy();

    doDestroy();

    verifyDestroyReturnedTrue();
  }

  @Test
  public void expireDestroyOfEntryInUseIsCancelled()
      throws RegionClearedException {
    givenConcurrencyChecks(true);
    givenExistingEntry();
    givenEntryIsInUseByTransaction();
    givenExpireDestroy();

    doDestroy();

    verifyDestroyReturnedFalse();
    verifyNoDestroyInvocationsOnEntry(existingRegionEntry);
    verifyNoDestroyInvocationsOnRegion();
  }


  @Test
  public void destroyOfExistingEntryCallsUpdateSizeOnRemove() {
    givenConcurrencyChecks(false);
    givenExistingEntry();

    doDestroy();

    verifyDestroyReturnedTrue();
    verify(owner, times(1)).updateSizeOnRemove(any(), anyInt());
  }

  /**
   * This might be a bug. It seems like we should have created a tombstone but we have no version
   * tag so that might be the cause of this bug.
   */
  @Test
  public void destroyOfExistingEntryWithConcurrencyChecksAndNoVersionTagDestroysWithoutTombstone()
      throws Exception {
    givenConcurrencyChecks(true);
    givenExistingEntry();
    givenExistingEntryWithNoVersionStamp();

    doDestroy();

    verifyDestroyReturnedTrue();
    verifyMapDoesNotContainKey();
    verify(existingRegionEntry, times(1)).removePhase2();
    verifyInvokedDestroyMethodsOnRegion(false);
  }

  @Test
  public void destroyOfExistingEntryWithConcurrencyChecksAddsTombstone() throws Exception {
    givenConcurrencyChecks(true);
    givenExistingEntryWithVersionTag();

    doDestroy();

    verifyDestroyReturnedTrue();
    verifyEntryDestroyed(existingRegionEntry, false);
    verifyInvokedDestroyMethodsOnRegion(false);
  }

  @Test
  public void evictDestroyOfExistingEntryWithConcurrencyChecksAddsTombstone() throws Exception {
    givenConcurrencyChecks(true);
    givenEviction();
    givenExistingEntryWithVersionTag();

    doDestroy();

    verifyDestroyReturnedTrue();
    verifyEntryDestroyed(existingRegionEntry, false);
    verifyInvokedDestroyMethodsOnRegion(false);
  }

  @Test
  public void destroyWithEmptyRegionWithConcurrencyChecksThrowsException() {
    givenConcurrencyChecks(true);

    doDestroy();

    verify(owner, times(1)).checkEntryNotFound(any());
  }

  @Test
  public void destroyWithEmptyNonReplicateRegionWithConcurrencyChecksAndRemoteEventThrowsException() {
    givenConcurrencyChecks(true);
    givenEmptyDataPolicy();
    givenOriginIsRemote();

    doDestroy();

    verify(owner, times(1)).checkEntryNotFound(any());
  }

  @Test
  public void destroyWithEmptyNonReplicateRegionWithConcurrencyChecksAndRemoteEventAndCacheWriteThrowsException() {
    givenConcurrencyChecks(true);
