/*
 * 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; private 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; private software distributed under the
 * License
 * is distributed on an "AS IS" BASIS; private WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND; private
 * either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.internal.cache;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.SoftAssertions.assertSoftly;
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.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.same;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.mockito.MockitoAnnotations.initMocks;

import java.util.ArrayList;
import java.util.List;

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

import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.query.internal.index.IndexManager;
import org.apache.geode.cache.query.internal.index.IndexProtocol;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.TXEntryState.DistTxThinEntryState;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.versions.VersionStamp;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.util.concurrent.ConcurrentMapWithReusableEntries;

@SuppressWarnings("unchecked")
public class AbstractRegionMapTxApplyDestroyTest {
  // parameters
  private final Object key = "key";
  @Mock
  private TXId txId;
  private TXRmtEvent txEvent;
  private boolean inTokenMode;
  private boolean inRI;
  private final Operation operation = Operation.DESTROY;
  @Mock
  private EventID eventId;
  private final Object aCallbackArgument = "aCallbackArgument";
  private final List<EntryEventImpl> pendingCallbacks = new ArrayList<>();
  private final FilterRoutingInfo filterRoutingInfo = null; // Provide a meaningful value for this?
  @Mock
  private ClientProxyMembershipID bridgeContext;
  private final boolean isOriginRemote = false;
  @Mock
  private TXEntryState txEntryState;
  private VersionTag versionTag;
  private long tailKey = 223L;

  @Mock
  private InternalDistributedMember myId;
  @Mock
  private InternalDistributedMember remoteId;
  @Mock
  private KeyInfo keyInfo;
  @Mock
  private ConcurrentMapWithReusableEntries entryMap;
  @Mock
  private RegionEntryFactory regionEntryFactory;
  @Mock
  private RegionEntry existingRegionEntry;
  @Mock
  private RegionEntry factoryRegionEntry;
  @Mock
  private RegionEntry oldRegionEntry;
  @Mock
  private PartitionedRegion partitionedRegion;
  @Mock
  private VersionTag existingVersionTag;
  @Mock
  private CachePerfStats cachePerfStats;

  private LocalRegion owner;
  private TestableAbstractRegionMap regionMap;

  @Before
  public void setup() {
    initMocks(this);
    VersionStamp versionStamp = mock(VersionStamp.class);
    when(versionStamp.asVersionTag()).thenReturn(existingVersionTag);
    when(existingRegionEntry.getVersionStamp()).thenReturn(versionStamp);
    when(existingRegionEntry.getKey()).thenReturn(key);
    when(factoryRegionEntry.getKey()).thenReturn(key);
    when(oldRegionEntry.getKey()).thenReturn(key);
    when(keyInfo.getKey()).thenReturn(key);
    when(txId.getMemberId()).thenReturn(myId);
  }

  // tests for no region entry.
  // Each of these tests requires givenNotInTokenMode and givenNoConcurrencyChecks

  @Test
  public void txApplyDestroySetCorrectPendingCallback_givenNoRegionEntryNotInTokenModeNoConcurrencyChecks() {
    givenLocalRegion();
    givenNoRegionEntry();
    givenNotInTokenMode();
    givenNoConcurrencyChecks();
    when(owner.generateEventID()).thenReturn(true);
    when(keyInfo.getCallbackArg()).thenReturn(aCallbackArgument);

    doTxApplyDestroy();
    assertThat(pendingCallbacks).hasSize(1);
    EntryEventImpl callbackEvent = pendingCallbacks.get(0);

    assertSoftly(softly -> {
      softly.assertThat(callbackEvent.getRegion()).isSameAs(owner);
      softly.assertThat(callbackEvent.getOperation()).isSameAs(operation);
      softly.assertThat(callbackEvent.getKey()).isSameAs(key);
      softly.assertThat(callbackEvent.getNewValue()).isNull();
      softly.assertThat(callbackEvent.getTransactionId()).isSameAs(txId);
      softly.assertThat(callbackEvent.getEventId()).isSameAs(eventId);
      softly.assertThat(callbackEvent.getCallbackArgument()).isSameAs(aCallbackArgument);
      softly.assertThat(callbackEvent.getLocalFilterInfo()).isSameAs(filterRoutingInfo);
      softly.assertThat(callbackEvent.getContext()).isSameAs(bridgeContext);
      softly.assertThat(callbackEvent.isOriginRemote()).isEqualTo(isOriginRemote);
      softly.assertThat(callbackEvent.getVersionTag()).isEqualTo(versionTag);
      softly.assertThat(callbackEvent.getTailKey()).isEqualTo(tailKey);
    });
  }

  @Test
  public void addsCallbackEvent_givenNoRegionEntryNotInTokenModeNoConcurrencyChecks_andBucket() {
    givenBucketRegion();
    givenNoRegionEntry();
    givenNotInTokenMode();
    givenNoConcurrencyChecks();

    when(partitionedRegion.generateEventID()).thenReturn(true);
    when(keyInfo.getCallbackArg()).thenReturn(aCallbackArgument);

    doTxApplyDestroy();

    verify(owner).txHandleWANEvent(any());
    verify(txEntryState).setTailKey(tailKey);

    assertThat(pendingCallbacks).hasSize(1);
    EntryEventImpl callbackEvent = pendingCallbacks.get(0);

    assertSoftly(softly -> {
      softly.assertThat(callbackEvent.getRegion()).isSameAs(partitionedRegion);
      softly.assertThat(callbackEvent.getOperation()).isSameAs(operation);
      softly.assertThat(callbackEvent.getKey()).isSameAs(key);
      softly.assertThat(callbackEvent.getNewValue()).isNull();
      softly.assertThat(callbackEvent.getTransactionId()).isSameAs(txId);
      softly.assertThat(callbackEvent.getEventId()).isSameAs(eventId);
      softly.assertThat(callbackEvent.getCallbackArgument()).isSameAs(aCallbackArgument);
      softly.assertThat(callbackEvent.getLocalFilterInfo()).isSameAs(filterRoutingInfo);
      softly.assertThat(callbackEvent.getContext()).isSameAs(bridgeContext);
      softly.assertThat(callbackEvent.isOriginRemote()).isEqualTo(isOriginRemote);
      softly.assertThat(callbackEvent.getVersionTag()).isEqualTo(versionTag);
      softly.assertThat(callbackEvent.getTailKey()).isEqualTo(tailKey);
    });
  }

  @Test
  public void txApplyDestroyCallReleaseEvent_givenNoRegionEntryNotInTokenModeNoConcurrencyChecksAndBucket_whenHandleWANEventThrows() {
    givenBucketRegion();
    givenNoRegionEntry();
    givenNotInTokenMode();
    givenNoConcurrencyChecks();
    doThrow(RuntimeException.class).when(owner).txHandleWANEvent(any());

    assertThatThrownBy(this::doTxApplyDestroy).isInstanceOf(RuntimeException.class);

    verify(regionMap, times(1)).releaseEvent(any());
  }

  // tests for "existingRegionEntry"

  @Test
  public void txApplyDestroyHasNoPendingCallback_givenExistingRegionEntryThatIsRemoved() {
    givenLocalRegion();
    givenExistingRegionEntry();
    when(existingRegionEntry.isRemoved()).thenReturn(true);

    doTxApplyDestroy();

    assertThat(pendingCallbacks).isEmpty();
  }

  @Test
  public void txApplyDestroyInvokesRescheduleTombstone_givenExistingRegionEntryThatIsTombstone() {
    givenLocalRegion();
    givenExistingRegionEntry();
    when(existingRegionEntry.isRemoved()).thenReturn(true);
    when(existingRegionEntry.isTombstone()).thenReturn(true);

    doTxApplyDestroy();

    verify(owner, times(1)).rescheduleTombstone(same(existingRegionEntry), any());
  }

  @Test
  public void doesNotAddCallbackEvent_ifExistingRegionEntryIsTombstone() throws Exception {
    givenLocalRegion();
    givenExistingRegionEntry();
    when(existingRegionEntry.isRemoved()).thenReturn(true);
    when(existingRegionEntry.isTombstone()).thenReturn(true);

    doTxApplyDestroy();

    assertThat(pendingCallbacks).isEmpty();
    verify(existingRegionEntry, never()).makeTombstone(any(), any());
    verify(txEntryState, never()).setVersionTag(any());
  }

  @Test
  public void txApplyDestroyHasNoPendingCallback_givenExistingRegionEntryWithInTokenModeAndNotInRI() {
    givenLocalRegion();
    givenNoConcurrencyChecks();
    givenExistingRegionEntry();
    inTokenMode = true;
    inRI = false;

    doTxApplyDestroy();

    assertThat(pendingCallbacks).isEmpty();
    verify(owner, times(1)).txApplyDestroyPart2(same(existingRegionEntry), eq(key),
        eq(inTokenMode), eq(false), eq(false));
  }

  @Test
  public void doesNotAddCallbackEvent_givenExistingRegionEntryWithInTokenModeAndNotInRI()
      throws Exception {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenExistingRegionEntry();
    inTokenMode = true;
    inRI = false;

    doTxApplyDestroy();

    assertThat(pendingCallbacks).isEmpty();
    verify(owner).generateAndSetVersionTag(any(EntryEventImpl.class), same(existingRegionEntry));
    verify(existingRegionEntry, never()).makeTombstone(any(), any());
    verify(txEntryState).setVersionTag(any());
  }

  @Test
  public void setsRegionEntryOnEvent_ifExistingRegionEntryIsValid() throws Exception {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenExistingRegionEntry();

    Object oldValue = "oldValue";
    when(existingRegionEntry.getValueInVM(owner)).thenReturn(oldValue);

    doTxApplyDestroy();

    assertThat(pendingCallbacks).size().isEqualTo(1);
    EntryEventImpl event = pendingCallbacks.get(0);
    assertThat(event.getRegionEntry()).isSameAs(existingRegionEntry);
    assertThat(event.getOldValue()).isSameAs(oldValue);

    verify(owner).generateAndSetVersionTag(any(EntryEventImpl.class), same(existingRegionEntry));
    verify(existingRegionEntry, never()).makeTombstone(any(), any());
    verify(txEntryState).setVersionTag(event.getVersionTag());
  }

  @Test
  public void txApplyDestroyUpdateIndexes_givenExistingRegionEntryThatIsValid() throws Exception {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenExistingRegionEntry();
    IndexManager indexManager = mock(IndexManager.class);
    when(owner.getIndexManager()).thenReturn(indexManager);

    doTxApplyDestroy();

    verify(indexManager, times(1)).updateIndexes(same(existingRegionEntry),
        eq(IndexManager.REMOVE_ENTRY), eq(IndexProtocol.OTHER_OP));
  }

  @Test
  public void txApplyDestroyCallsAddDestroy_givenExistingRegionEntryThatIsValidAndTxRmtEvent() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenExistingRegionEntry();
    txEvent = mock(TXRmtEvent.class);

    doTxApplyDestroy();

    verify(txEvent, times(1)).addDestroy(same(owner), same(existingRegionEntry), eq(key),
        same(aCallbackArgument));
  }

  @Test
  public void callsProcessAndGenerateTXVersionTag_givenExistingRegionEntryThatIsValid() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenExistingRegionEntry();

    doTxApplyDestroy();

    EntryEventImpl callbackEvent = pendingCallbacks.get(0);
    verify(regionMap, times(1)).processAndGenerateTXVersionTag(same(callbackEvent),
        same(existingRegionEntry), same(txEntryState));
    assertThat(callbackEvent.getNextRegionVersion()).isEqualTo(-1L); // Default value
  }

  @Test
  public void setsEventNextRegionVersionOnCallbackEvent_givenExistingRegionEntryThatIsValidAndDistTxEntryStateExists() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenExistingRegionEntry();

    DistTxThinEntryState distTxEntryStates = mock(DistTxThinEntryState.class);
    when(txEntryState.getDistTxEntryStates()).thenReturn(distTxEntryStates);
    when(distTxEntryStates.getRegionVersion()).thenReturn(999L);

    doTxApplyDestroy();

    EntryEventImpl callbackEvent = pendingCallbacks.get(0);
    assertThat(callbackEvent.getNextRegionVersion()).isEqualTo(999L); // Default value
  }

  @Test
  public void txApplyDestroySetsValueToDestroyToken_givenExistingRegionEntryThatIsValidWithInTokenMode()
      throws Exception {
    givenLocalRegion();
    givenNoConcurrencyChecks();
    givenExistingRegionEntry();
    inTokenMode = true;

    doTxApplyDestroy();

    verify(existingRegionEntry, times(1)).setValue(same(owner), eq(Token.DESTROYED));
  }

  @Test
  public void txApplyDestroyHandlesClear_givenExistingRegionEntryThatIsValidWithInTokenModeAndSetValueThrowsRegionClearedException()
      throws Exception {
    givenLocalRegion();
    givenNoConcurrencyChecks();
    givenExistingRegionEntry();
    inTokenMode = true;
    doThrow(RegionClearedException.class).when(existingRegionEntry).setValue(same(owner),
        eq(Token.DESTROYED));

    doTxApplyDestroy();

    verify(owner, times(1)).txApplyDestroyPart2(any(), any(), anyBoolean(), eq(true), anyBoolean());
    verify(regionMap, never()).lruEntryDestroy(any());
  }

  @Test
  public void txApplyDestroyHandlesNoClear_givenExistingRegionEntryThatIsValidWithInTokenMode() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenExistingRegionEntry();
    inTokenMode = true;

    doTxApplyDestroy();

    verify(owner, times(1)).txApplyDestroyPart2(any(), any(), anyBoolean(), eq(false),
        anyBoolean());
    verify(regionMap, times(1)).lruEntryDestroy(any());
  }

  @Test
  public void txApplyDestroyCallsUnscheduleTombstone_givenExistingRegionEntryThatIsTombstoneWithInTokenMode() {
    givenLocalRegion();
    givenNoConcurrencyChecks();
    givenExistingRegionEntry();
    inTokenMode = true;
    when(existingRegionEntry.getValueInVM(owner)).thenReturn(Token.TOMBSTONE);

    doTxApplyDestroy();

    verify(owner, times(1)).unscheduleTombstone(same(existingRegionEntry));
  }

  @Test
  public void txApplyDestroyCallsRemoveEntry_givenExistingRegionEntryThatIsValidWithoutInTokenModeWithoutConcurrencyCheck()
      throws Exception {
    givenLocalRegion();
    givenNoConcurrencyChecks();
    givenExistingRegionEntry();
    inTokenMode = false;

    doTxApplyDestroy();

    verify(existingRegionEntry, times(1)).removePhase1(same(owner), eq(false));
    verify(existingRegionEntry, times(1)).removePhase2();
    verify(regionMap, times(1)).removeEntry(eq(key), same(existingRegionEntry), eq(false));
  }

  @Test
  public void txApplyDestroyCallsRescheduleTombstone_givenExistingRegionEntryThatIsValidWithoutInTokenModeWithConcurrencyCheckButNoVersionTag()
      throws Exception {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenExistingRegionEntry();
    inTokenMode = false;
    versionTag = null;

    doTxApplyDestroy();

    verify(existingRegionEntry, times(1)).removePhase1(same(owner), eq(false));
    verify(existingRegionEntry, times(1)).removePhase2();
    verify(regionMap, times(1)).removeEntry(eq(key), same(existingRegionEntry), eq(false));
  }

  @Test
  public void txApplyDestroyCallsMakeTombstone_givenExistingRegionEntryThatIsValidWithoutInTokenModeWithConcurrencyCheckAndVersionTag()
      throws Exception {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenExistingRegionEntry();
    inTokenMode = false;
    versionTag = mock(VersionTag.class);

    doTxApplyDestroy();

    verify(existingRegionEntry, times(1)).makeTombstone(same(owner), same(versionTag));
  }

  @Test
  public void txApplyDestroyCallsUpdateSizeOnRemove_givenExistingRegionEntryThatIsValid() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenExistingRegionEntry();
    int oldSize = 79;
    when(owner.calculateRegionEntryValueSize(same(existingRegionEntry))).thenReturn(oldSize);

    doTxApplyDestroy();

    verify(owner, times(1)).updateSizeOnRemove(eq(key), eq(oldSize));
  }

  @Test
  public void txApplyDestroyCallsReleaseEvent_givenExistingRegionEntry() {
    givenLocalRegion();
    givenExistingRegionEntry();

    doTxApplyDestroy();

    verify(regionMap, times(1)).releaseEvent(any());
  }

  @Test
  public void txApplyDestroyHasPendingCallback_givenExistingRegionEntryWithoutInTokenModeAndNotInRI() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenExistingRegionEntry();
    inTokenMode = false;
    inRI = false;

    doTxApplyDestroy();

    assertThat(pendingCallbacks).hasSize(1);
    verify(owner, times(1)).txApplyDestroyPart2(same(existingRegionEntry), eq(key), eq(inTokenMode),
        eq(false), eq(false));

  }

  @Test
  public void txApplyDestroyHasPendingCallback_givenExistingRemovedRegionEntryWithoutInTokenModeAndNotInRI() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenExistingRemovedRegionEntry();
    inTokenMode = false;
    inRI = false;

    doTxApplyDestroy();

    assertThat(pendingCallbacks).hasSize(1);
    verify(owner, times(1)).txApplyDestroyPart2(same(existingRegionEntry), eq(key), eq(inTokenMode),
        eq(false), eq(true));

  }

  @Test
  public void txApplyDestroyHasPendingCallback_givenExistingRegionEntryWithoutInTokenModeAndWithInRI() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenExistingRegionEntry();
    inTokenMode = false;
    inRI = true;

    doTxApplyDestroy();

    assertThat(pendingCallbacks).hasSize(1);
    verify(owner, times(1)).txApplyDestroyPart2(same(existingRegionEntry), eq(key), eq(inTokenMode),
        eq(false), eq(false));
  }

  @Test
  public void txApplyDestroyHasPendingCallback_givenExistingRemovedRegionEntryWithoutInTokenModeAndWithInRI() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenExistingRemovedRegionEntry();
    inTokenMode = false;
    inRI = true;

    doTxApplyDestroy();

    assertThat(pendingCallbacks).hasSize(1);
    verify(owner, times(1)).txApplyDestroyPart2(same(existingRegionEntry), eq(key), eq(inTokenMode),
        eq(false), eq(true));
  }

  @Test
  public void txApplyDestroyHasPendingCallback_givenExistingRegionEntryWithInTokenModeAndInRI() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenExistingRegionEntry();
    inTokenMode = true;
    inRI = true;

    doTxApplyDestroy();

    assertThat(pendingCallbacks).hasSize(1);
    verify(owner, times(1)).txApplyDestroyPart2(same(existingRegionEntry), eq(key),
        eq(!inTokenMode),
        eq(false), eq(false));
  }

  @Test
  public void txApplyDestroyHasPendingCallback_givenExistingRemovedRegionEntryWithInTokenModeAndInRI() {
    givenLocalRegion();
    givenNoConcurrencyChecks();
    givenExistingRemovedRegionEntry();
    inTokenMode = true;
    inRI = true;

    doTxApplyDestroy();

    verify(owner, times(1)).txApplyDestroyPart2(same(existingRegionEntry), eq(key), eq(inTokenMode),
        eq(false), eq(true));
  }

  @Test
  public void txApplyDestroyHasPendingCallback_givenExistingRemovedRegionEntryWithInTokenModeAndInRIWithConcurrencyCheck() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenExistingRemovedRegionEntry();
    inTokenMode = true;
    inRI = true;

    doTxApplyDestroy();

    assertThat(pendingCallbacks).hasSize(1);
    verify(owner, times(1)).txApplyDestroyPart2(same(existingRegionEntry), eq(key),
        eq(!inTokenMode),
        eq(false), eq(true));
  }

  @Test
  public void txApplyDestroyHasNoPendingCallback_givenExistingRegionEntryWithPartitionedRegion() {
    givenBucketRegion();
    givenExistingRegionEntry();

    doTxApplyDestroy();

    assertThat(pendingCallbacks).isEmpty();
    verify(owner, times(1)).txApplyDestroyPart2(same(existingRegionEntry), eq(key), eq(inTokenMode),
        eq(false), eq(false));
  }

  @Test
  public void txApplyDestroyHasNoPendingCallback_givenExistingRemovedRegionEntryWithPartitionedRegion() {
    givenBucketRegion();
    givenExistingRemovedRegionEntry();

    doTxApplyDestroy();

    assertThat(pendingCallbacks).isEmpty();
    verify(owner, times(1)).txApplyDestroyPart2(same(existingRegionEntry), eq(key), eq(inTokenMode),
        eq(false), eq(true));
  }

  @Test
  public void txApplyDestroyCallsTxHandleWanEvent_givenExistingRegionEntryWithPartitionedRegion() {
    givenBucketRegion();
    givenExistingRegionEntry();

    doTxApplyDestroy();

    verify(owner).txHandleWANEvent(any());
    verify(txEntryState).setTailKey(tailKey);
  }

  @Test
  public void txApplyDestroyDoesNotSetTailKey_givenTailKeyIsNegativeOne() {
    givenBucketRegion();
    givenExistingRegionEntry();
    tailKey = -1;

    doTxApplyDestroy();

    verify(owner).txHandleWANEvent(any());
    verify(txEntryState, never()).setTailKey(tailKey);
  }

  @Test
  public void txApplyDestroyDoesNotCallSetVersionTag_givenExistingRegionEntryWithPartitionedRegionButNoConcurrencyChecks() {
    givenBucketRegion();
    givenExistingRegionEntry();
    versionTag = mock(VersionTag.class);
    givenNoConcurrencyChecks();

    doTxApplyDestroy();

    verify(txEntryState, never()).setVersionTag(any());
  }

  @Test
  public void txApplyDestroyPreparesAndReleasesIndexManager_givenExistingRegionEntryWithIndexManager() {
    givenLocalRegion();
    givenExistingRegionEntry();
    IndexManager indexManager = mock(IndexManager.class);
    when(owner.getIndexManager()).thenReturn(indexManager);

    doTxApplyDestroy();

    InOrder inOrder = inOrder(indexManager);
    inOrder.verify(indexManager, times(1)).waitForIndexInit();
    inOrder.verify(indexManager, times(1)).countDownIndexUpdaters();
  }

  @Test
  public void txApplyDestroyCallsSetVersionTag_givenExistingRegionEntryWithPartitionedRegionAndConcurrencyChecks() {
    givenBucketRegion();
    givenExistingRegionEntry();
    versionTag = mock(VersionTag.class);
    givenConcurrencyChecks();

    doTxApplyDestroy();

    verify(txEntryState, times(1)).setVersionTag(same(versionTag));
  }

  @Test
  public void txApplyDestroyHasNoPendingCallback_givenExistingRegionEntryWithoutConcurrencyChecks() {
    givenLocalRegion();
    givenExistingRegionEntry();
    givenNoConcurrencyChecks();

    doTxApplyDestroy();

    assertThat(pendingCallbacks).isEmpty();
    verify(owner, times(1)).txApplyDestroyPart2(same(existingRegionEntry), eq(key), eq(inTokenMode),
        eq(false), eq(false));
  }

  @Test
  public void txApplyDestroyHasNoPendingCallback_givenExistingRemovedRegionEntryWithoutConcurrencyChecks() {
    givenLocalRegion();
    givenExistingRemovedRegionEntry();
    givenNoConcurrencyChecks();

    doTxApplyDestroy();

    assertThat(pendingCallbacks).isEmpty();
    verify(owner, times(1)).txApplyDestroyPart2(same(existingRegionEntry), eq(key), eq(inTokenMode),
        eq(false), eq(true));
  }

  @Test
  public void txApplyDestroyHasPendingCallback_givenExistingRegionEntryWithShouldDispatchListenerEvent() {
    givenLocalRegion();
    givenExistingRegionEntry();
    when(owner.shouldDispatchListenerEvent()).thenReturn(true);

    doTxApplyDestroy();

    assertThat(pendingCallbacks).hasSize(1);
    verify(owner, times(1)).txApplyDestroyPart2(same(existingRegionEntry), eq(key), eq(inTokenMode),
        eq(false), eq(false));
  }

  @Test
  public void txApplyDestroyHasPendingCallback_givenExistingRemovedRegionEntryWithShouldDispatchListenerEvent() {
    givenLocalRegion();
    givenExistingRemovedRegionEntry();
    when(owner.shouldDispatchListenerEvent()).thenReturn(true);

    doTxApplyDestroy();

    assertThat(pendingCallbacks).hasSize(1);
    verify(owner, times(1)).txApplyDestroyPart2(same(existingRegionEntry), eq(key), eq(inTokenMode),
        eq(false), eq(true));
  }

  @Test
  public void txApplyDestroyHasPendingCallback_givenExistingRegionEntryWithshouldNotifyBridgeClients() {
    givenLocalRegion();
    givenExistingRegionEntry();
    when(owner.shouldNotifyBridgeClients()).thenReturn(true);

    doTxApplyDestroy();

    assertThat(pendingCallbacks).hasSize(1);
    verify(owner, times(1)).txApplyDestroyPart2(same(existingRegionEntry), eq(key), eq(inTokenMode),
        eq(false), eq(false));
  }

  @Test
  public void txApplyDestroyHasPendingCallback_givenExistingRemovedRegionEntryWithshouldNotifyBridgeClients() {
    givenLocalRegion();
    givenExistingRemovedRegionEntry();
    when(owner.shouldNotifyBridgeClients()).thenReturn(true);

    doTxApplyDestroy();

    assertThat(pendingCallbacks).hasSize(1);
    verify(owner, times(1)).txApplyDestroyPart2(same(existingRegionEntry), eq(key), eq(inTokenMode),
        eq(false), eq(true));
  }

  @Test
  public void txApplyDestroyDoesCallTxApplyDestroyPart2_givenFactoryRegionEntry() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenFactoryRegionEntry();
    versionTag = mock(VersionTag.class);

    doTxApplyDestroy();

    verify(owner, times(1)).txApplyDestroyPart2(same(factoryRegionEntry), eq(key), eq(inTokenMode),
        eq(false), eq(true));
  }

  // tests for "factoryRegionEntry" (that is, no existing region entry and no oldRegionEntry).
  // All these tests require no existing region entry (that is, not found by getEntry).
  // All these tests need one of the following: givenConcurrencyChecks OR inTokenMode

  @Test
  public void txApplyDestroyCallsCreateEntry_givenFactoryRegionEntryAndInTokenMode() {
    givenLocalRegion();
    givenFactoryRegionEntry();
    inTokenMode = true;

    doTxApplyDestroy();

    verify(regionEntryFactory, times(1)).createEntry(same(owner), eq(key),
        eq(Token.REMOVED_PHASE1));
  }

  @Test
  public void txApplyDestroyCallsCreateEntry_givenFactoryRegionEntryAndConcurrencyChecks() {
    givenLocalRegion();
    givenFactoryRegionEntry();
    givenConcurrencyChecks();

    doTxApplyDestroy();

    verify(regionEntryFactory, times(1)).createEntry(same(owner), eq(key),
        eq(Token.REMOVED_PHASE1));
  }

  @Test
  public void txApplyDestroyPreparesAndReleasesIndexManager_givenFactoryRegionEntryAndConcurrencyChecksWithIndexManager() {
    givenLocalRegion();
    givenFactoryRegionEntry();
    givenConcurrencyChecks();
    IndexManager indexManager = mock(IndexManager.class);
    when(owner.getIndexManager()).thenReturn(indexManager);

    doTxApplyDestroy();

    InOrder inOrder = inOrder(indexManager);
    inOrder.verify(indexManager, times(1)).waitForIndexInit();
    inOrder.verify(indexManager, times(1)).countDownIndexUpdaters();
  }

  @Test
  public void txApplyDestroyHasNoPendingCallback_givenFactoryRegionEntryWithInTokenModeAndNotInRI() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenFactoryRegionEntry();
    inTokenMode = true;
    inRI = false;

    doTxApplyDestroy();

    assertThat(pendingCallbacks).isEmpty();
  }

  @Test
  public void txApplyDestroyHasPendingCallback_givenFactoryRegionEntryWithoutInTokenModeAndNotInRI() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenFactoryRegionEntry();
    inTokenMode = false;
    inRI = false;

    doTxApplyDestroy();

    assertThat(pendingCallbacks).hasSize(1);
  }

  @Test
  public void txApplyDestroyHasPendingCallback_givenFactoryRegionEntryWithoutInTokenModeAndWithInRI() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenFactoryRegionEntry();
    inTokenMode = false;
    inRI = true;

    doTxApplyDestroy();

    assertThat(pendingCallbacks).hasSize(1);
  }

  @Test
  public void txApplyDestroyHasPendingCallback_givenFactoryRegionEntryWithInTokenModeAndInRI() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenFactoryRegionEntry();
    inTokenMode = true;
    inRI = true;

    doTxApplyDestroy();

    assertThat(pendingCallbacks).hasSize(1);
  }

  @Test
  public void txApplyDestroyHasNoPendingCallback_givenFactoryRegionEntryWithPartitionedRegion() {
    givenBucketRegion();
    givenConcurrencyChecks();
    givenFactoryRegionEntry();
    when(partitionedRegion.getConcurrencyChecksEnabled()).thenReturn(false);

    doTxApplyDestroy();

    assertThat(pendingCallbacks).isEmpty();
  }

  @Test
  public void txApplyDestroyHasNoPendingCallback_givenFactoryRegionEntryWithoutConcurrencyChecksInTokenMode() {
    givenLocalRegion();
    givenFactoryRegionEntry();
    givenNoConcurrencyChecks();
    inTokenMode = true;

    doTxApplyDestroy();

    assertThat(pendingCallbacks).isEmpty();
  }

  @Test
  public void txApplyDestroyHasPendingCallback_givenFactoryRegionEntryWithShouldDispatchListenerEvent() {
    givenLocalRegion();
    givenNoConcurrencyChecks();
    givenFactoryRegionEntry();
    inTokenMode = true;
    inRI = true;
    when(owner.shouldDispatchListenerEvent()).thenReturn(true);

    doTxApplyDestroy();

    assertThat(pendingCallbacks).hasSize(1);
  }

  @Test
  public void txApplyDestroyHasPendingCallback_givenFactoryRegionEntryWithshouldNotifyBridgeClients() {
    givenLocalRegion();
    givenNoConcurrencyChecks();
    givenFactoryRegionEntry();
    inTokenMode = true;
    inRI = true;
    when(owner.shouldNotifyBridgeClients()).thenReturn(true);

    doTxApplyDestroy();

    assertThat(pendingCallbacks).hasSize(1);
  }

  @Test
  public void txApplyDestroySetsRegionEntryOnEvent_givenFactoryRegionEntry() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenFactoryRegionEntry();

    doTxApplyDestroy();

    EntryEventImpl event = pendingCallbacks.get(0);
    assertThat(event.getRegionEntry()).isSameAs(factoryRegionEntry);
  }

  @Test
  public void txApplyDestroySetsOldValueOnEventToNotAvailable_givenFactoryRegionEntry() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenFactoryRegionEntry();

    doTxApplyDestroy();

    EntryEventImpl event = pendingCallbacks.get(0);
    assertThat(event.getRawOldValue()).isSameAs(Token.NOT_AVAILABLE);
  }

  @Test
  public void txApplyDestroyCallsTxHandleWanEvent_givenFactoryRegionEntryWithPartitionedRegion() {
    givenBucketRegion();
    givenConcurrencyChecks();
    givenFactoryRegionEntry();

    doTxApplyDestroy();

    verify(owner).txHandleWANEvent(any());
    verify(txEntryState).setTailKey(tailKey);
  }

  @Test
  public void txApplyDestroyCallsProcessAndGenerateTXVersionTag_givenFactoryRegionEntry() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenFactoryRegionEntry();

    doTxApplyDestroy();

    verify(regionMap, times(1)).processAndGenerateTXVersionTag(any(), same(factoryRegionEntry),
        same(txEntryState));
  }

  @Test
  public void txApplyDestroySetsRemoteOrigin_givenFactoryRegionEntryAndRemoteTxId() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenFactoryRegionEntry();
    when(txId.getMemberId()).thenReturn(remoteId);

    doTxApplyDestroy();

    EntryEventImpl event = pendingCallbacks.get(0);
    assertThat(event.isOriginRemote()).isTrue();
  }

  @Test
  public void txApplyDestroySetsRemoteOrigin_givenFactoryRegionEntry() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenFactoryRegionEntry();

    doTxApplyDestroy();

    EntryEventImpl event = pendingCallbacks.get(0);
    assertThat(event.isOriginRemote()).isFalse();
  }

  @Test
  public void txApplyDestroySetsRegionOnEvent_givenFactoryRegionEntryAndBucket() {
    givenBucketRegion();
    givenConcurrencyChecks();
    givenFactoryRegionEntry();

    doTxApplyDestroy();

    EntryEventImpl event = pendingCallbacks.get(0);
    assertThat(event.getRegion()).isSameAs(partitionedRegion);
  }

  @Test
  public void txApplyDestroyNeverCallsUpdateSizeOnCreate_givenFactoryRegionEntry() {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenFactoryRegionEntry();

    doTxApplyDestroy();

    verify(owner, never()).updateSizeOnCreate(any(), anyInt());
  }

  @Test
  public void txApplyDestroyCallsMakeTombstone_givenFactoryRegionEntryWithConcurrencyCheckAndVersionTag()
      throws Exception {
    givenLocalRegion();
    givenConcurrencyChecks();
    givenFactoryRegionEntry();
    versionTag = mock(VersionTag.class);

    doTxApplyDestroy();

    verify(factoryRegionEntry, times(1)).makeTombstone(same(owner), same(versionTag));
  }

  @Test
