/*
 * 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;

import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Properties;
import java.util.Set;

import jakarta.transaction.Synchronization;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;

import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.AttributesMutator;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheEvent;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.CacheLoader;
import org.apache.geode.cache.CacheLoaderException;
import org.apache.geode.cache.CacheTransactionManager;
import org.apache.geode.cache.CacheWriter;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.CommitConflictException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.DiskStoreFactory;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.EntryExistsException;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.EvictionAction;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.FailedSynchronizationException;
import org.apache.geode.cache.LoaderHelper;
import org.apache.geode.cache.PartitionAttributes;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionEvent;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.TransactionEvent;
import org.apache.geode.cache.TransactionException;
import org.apache.geode.cache.TransactionId;
import org.apache.geode.cache.TransactionListener;
import org.apache.geode.cache.UnsupportedOperationInTransactionException;
import org.apache.geode.cache.query.Index;
import org.apache.geode.cache.query.IndexType;
import org.apache.geode.cache.query.Query;
import org.apache.geode.cache.query.QueryException;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.cache.query.SelectResults;
import org.apache.geode.cache.query.internal.index.IndexManager;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.cache.util.TransactionListenerAdapter;
import org.apache.geode.cache.util.TxEventTestUtil;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.internal.NanoTimer;
import org.apache.geode.internal.cache.AbstractRegion;
import org.apache.geode.internal.cache.CachePerfStats;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.NonTXEntry;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.TXManagerImpl;
import org.apache.geode.internal.cache.TXStateProxy;
import org.apache.geode.internal.util.StopWatch;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Tests basic transaction functionality
 *
 * @since GemFire 4.0
 */
@SuppressWarnings("deprecated")
public class TXJUnitTest {

  private int cbCount;
  private TransactionEvent te;

  protected int listenerAfterCommit;
  protected int listenerAfterFailedCommit;
  protected int listenerAfterRollback;
  protected int listenerClose;
  protected CacheTransactionManager txMgr;

  protected GemFireCacheImpl cache;
  protected Region<String, String> region;

  @Rule
  public TestName testName = new TestName();

  private boolean isPR() {
    return (region instanceof PartitionedRegion);
  }

  protected void createCache() throws Exception {
    Properties properties = new Properties();
    properties.setProperty(MCAST_PORT, "0"); // loner

    cache = (GemFireCacheImpl) CacheFactory.create(DistributedSystem.connect(properties));

    createRegion();
    txMgr = cache.getCacheTransactionManager();
    listenerAfterCommit = 0;
    listenerAfterFailedCommit = 0;
    listenerAfterRollback = 0;
    listenerClose = 0;
  }

  protected void createRegion() throws Exception {
    region = createRegion(getClass().getSimpleName(), false);
  }

  protected Region createRegion(String regionName, boolean isConcurrencyChecksEnabled)
      throws Exception {
    AttributesFactory<String, String> attributesFactory = new AttributesFactory<>();
    attributesFactory
        .setDataPolicy(isConcurrencyChecksEnabled ? DataPolicy.REPLICATE : DataPolicy.NORMAL);
    attributesFactory
        .setScope(isConcurrencyChecksEnabled ? Scope.DISTRIBUTED_ACK : Scope.DISTRIBUTED_NO_ACK);
    attributesFactory.setConcurrencyChecksEnabled(isConcurrencyChecksEnabled); // test validation
                                                                               // expects this
                                                                               // behavior
    attributesFactory.setIndexMaintenanceSynchronous(true);

    return cache.createRegion(regionName, attributesFactory.create());
  }

  protected void closeCache() {
    if (cache != null) {
      if (txMgr != null) {
        try {
          txMgr.rollback();
        } catch (IllegalStateException ignore) {
        }
      }
      region = null;
      txMgr = null;
      Cache c = cache;
      cache = null;
      c.close();
    }
  }

  @Before
  public void setUpTXJUnitTest() throws Exception {
    createCache();
  }

  @After
  public void tearDownTXJUnitTest() throws Exception {
    closeCache();
  }

  @AfterClass
  public static void afterClass() {
    InternalDistributedSystem ids = InternalDistributedSystem.getAnyInstance();
    if (ids != null) {
      ids.disconnect();
    }
  }

  private void checkNoTxState() {
    assertEquals(null, txMgr.getTransactionId());
    assertTrue(!txMgr.exists());
    try {
      txMgr.commit();
      fail("expected IllegalStateException");
    } catch (CommitConflictException unexpected) {
      fail("did not expect " + unexpected);
    } catch (IllegalStateException ignored) {
    }
    try {
      txMgr.rollback();
      fail("expected IllegalStateException");
    } catch (IllegalStateException ignored) {
    }
  }

  @Test
  public void testSimpleOps() throws CacheException {
    final CachePerfStats stats = cache.getCachePerfStats();
    // See if things are ok when no transaction
    checkNoTxState();

    txMgr.begin();
    // now that a transaction exists make sure things behave as expected
    assertTrue(txMgr.getTransactionId() != null);
    assertTrue(txMgr.exists());
    try {
      txMgr.begin();
      fail("expected IllegalStateException");
    } catch (IllegalStateException ignored) {
    }
    try {
      txMgr.commit();
    } catch (CommitConflictException unexpected) {
      fail("did not expect " + unexpected);
    }
    checkNoTxState();

    txMgr.begin();
    txMgr.rollback();
    checkNoTxState();

    region.put("uaKey", "val");

    {
      Region.Entry cmtre = region.getEntry("uaKey");
      cmtre.setUserAttribute("uaValue1");
      assertEquals("uaValue1", cmtre.getUserAttribute());

      long txRollbackChanges = stats.getTxRollbackChanges();
      long txCommitChanges = stats.getTxCommitChanges();
      long txFailureChanges = stats.getTxFailureChanges();
      txMgr.begin();
      Region.Entry txre = region.getEntry("uaKey");
      assertEquals(region, txre.getRegion());
      if (isPR()) {
        region.put("1", "one");
        try {
          txre.setUserAttribute("uaValue2");
        } catch (UnsupportedOperationException e) {
          // expected
        }
        try {
          txre.getUserAttribute();
        } catch (UnsupportedOperationException e) {
          // expected
        }
      } else {
        assertEquals("uaValue1", txre.getUserAttribute());
        txre.setUserAttribute("uaValue2");
        assertEquals("uaValue2", txre.getUserAttribute());
      }
      txMgr.rollback();
      try {
        txre.getValue();
        fail("expected IllegalStateException");
      } catch (IllegalStateException ignored) {
      }
      try {
        txre.isDestroyed();
        fail("expected IllegalStateException");
      } catch (IllegalStateException ignored) {
      }
      try {
        txre.getUserAttribute();
        fail("expected IllegalStateException");
      } catch (IllegalStateException ignored) {
      }
      try {
        txre.setUserAttribute("foo");
        fail("expected IllegalStateException");
      } catch (IllegalStateException ignored) {
      }
      assertEquals(txRollbackChanges + 1, stats.getTxRollbackChanges());
      assertEquals(txCommitChanges, stats.getTxCommitChanges());
      assertEquals(txFailureChanges, stats.getTxFailureChanges());

      assertEquals("uaValue1", cmtre.getUserAttribute());
    }

    {
      long txRollbackChanges = stats.getTxRollbackChanges();
      long txCommitChanges = stats.getTxCommitChanges();
      long txFailureChanges = stats.getTxFailureChanges();
      region.create("key1", "value1");
      txMgr.begin();
      region.invalidate("key1");
      txMgr.rollback();
      assertEquals(region.get("key1"), "value1");
      assertEquals(txRollbackChanges + 1, stats.getTxRollbackChanges());
      assertEquals(txCommitChanges, stats.getTxCommitChanges());
      assertEquals(txFailureChanges, stats.getTxFailureChanges());

      txRollbackChanges = stats.getTxRollbackChanges();
      txMgr.begin();
      region.destroy("key1");
      txMgr.rollback();
      assertEquals(region.get("key1"), "value1");
      assertEquals(txRollbackChanges + 1, stats.getTxRollbackChanges());
      assertEquals(txCommitChanges, stats.getTxCommitChanges());
      assertEquals(txFailureChanges, stats.getTxFailureChanges());

      txRollbackChanges = stats.getTxRollbackChanges();
      txMgr.begin();
      region.put("key1", "value2");
      txMgr.rollback();
      assertEquals(region.get("key1"), "value1");
      assertEquals(txRollbackChanges + 1, stats.getTxRollbackChanges());
      assertEquals(txCommitChanges, stats.getTxCommitChanges());
      assertEquals(txFailureChanges, stats.getTxFailureChanges());
    }
  }

  @Test
  public void testWriteOps() throws CacheException {
    txMgr.begin();
    region.put("key1", "value1");
    region.put("key2", "value2");
    assertTrue(region.containsKey("key1"));
    assertTrue(region.containsValueForKey("key1"));
    assertEquals("key1", region.getEntry("key1").getKey());
    assertEquals("value1", region.getEntry("key1").getValue());
    assertEquals("value1", region.get("key1"));
    assertTrue(region.containsKey("key2"));
    assertTrue(region.containsValueForKey("key2"));
    assertEquals("key2", region.getEntry("key2").getKey());
    assertEquals("value2", region.getEntry("key2").getValue());
    assertEquals("value2", region.get("key2"));
    txMgr.rollback();
    assertTrue(!region.containsKey("key1"));
    assertTrue(!region.containsKey("key2"));

    txMgr.begin();
    region.create("key1", "value1");
    assertTrue(region.containsKey("key1"));
    assertTrue(region.containsValueForKey("key1"));
    assertEquals("key1", region.getEntry("key1").getKey());
    assertEquals("value1", region.getEntry("key1").getValue());
    assertEquals("value1", region.get("key1"));
    txMgr.rollback();
    assertTrue(!region.containsKey("key1"));

    region.create("key1", "value1");
    txMgr.begin();
    region.put("key1", "value2");
    txMgr.rollback();
    assertTrue(region.containsKey("key1"));
    assertEquals("value1", region.get("key1"));
    region.localDestroy("key1");

    region.create("key1", "value1");
    txMgr.begin();
    region.localDestroy("key1");
    assertTrue(!region.containsKey("key1"));
    assertTrue(!region.containsValueForKey("key1"));
    txMgr.rollback();
    assertTrue(region.containsKey("key1"));
    region.localDestroy("key1");

    region.create("key1", "value1");
    txMgr.begin();
    region.destroy("key1");
    assertTrue(!region.containsKey("key1"));
    assertTrue(!region.containsValueForKey("key1"));
    txMgr.rollback();
    assertTrue(region.containsKey("key1"));
    region.localDestroy("key1");

    region.create("key1", "value1");
    txMgr.begin();
    assertTrue(region.containsValueForKey("key1"));
    region.localInvalidate("key1");
    assertTrue(region.containsKey("key1"));
    assertTrue(!region.containsValueForKey("key1"));
    assertEquals(null, region.get("key1"));
    txMgr.rollback();
    assertTrue(region.containsValueForKey("key1"));
    region.localDestroy("key1");

    region.create("key1", "value1");
    txMgr.begin();
    assertTrue(region.containsValueForKey("key1"));
    region.invalidate("key1");
    assertTrue(region.containsKey("key1"));
    assertTrue(!region.containsValueForKey("key1"));
    assertEquals(null, region.get("key1"));
    txMgr.rollback();
    assertTrue(region.containsValueForKey("key1"));
    region.localDestroy("key1");

    // see if commits work
    assertTrue(!region.containsKey("key1"));
    assertTrue(!region.containsKey("key2"));
    txMgr.begin();
    region.create("key1", "value1");
    region.create("key2", "value2");
    txMgr.commit();
    assertTrue(region.containsKey("key1"));
    assertTrue(region.containsValueForKey("key1"));
    assertEquals("key1", region.getEntry("key1").getKey());
    assertEquals("value1", region.getEntry("key1").getValue());
    assertEquals("value1", region.get("key1"));
    assertTrue(region.containsKey("key2"));
    assertTrue(region.containsValueForKey("key2"));
    assertEquals("key2", region.getEntry("key2").getKey());
    assertEquals("value2", region.getEntry("key2").getValue());
    assertEquals("value2", region.get("key2"));
    region.localDestroy("key1");
    region.localDestroy("key2");
  }

  @Test
  public void testTwoRegionTxs() throws CacheException {
    final CachePerfStats stats = cache.getCachePerfStats();
    long txCommitChanges;
    TransactionId myTxId;

    AttributesFactory<String, String> attributesFactory = new AttributesFactory<>();
    attributesFactory.setScope(Scope.DISTRIBUTED_NO_ACK);

    Region<String, String> reg1 = region;
    Region<String, String> reg2 =
        cache.createRegion(getUniqueName(), attributesFactory.create());

    txMgr.setListener(new TransactionListener() {
      @Override
      public void afterCommit(TransactionEvent event) {
        listenerAfterCommit = 1;
        te = event;
      }

      @Override
      public void afterFailedCommit(TransactionEvent event) {
        listenerAfterFailedCommit = 1;
        te = event;
      }

      @Override
      public void afterRollback(TransactionEvent event) {
        listenerAfterRollback = 1;
        te = event;
      }

      @Override
      public void close() {
        listenerClose = 1;
      }
    });

    // see if commits work
    txCommitChanges = stats.getTxCommitChanges();
    assertTrue(!reg1.containsKey("key1"));
    assertTrue(!reg2.containsKey("key2"));
    txMgr.begin();
    myTxId = txMgr.getTransactionId();
    reg1.create("key1", "value1");
    reg2.create("key2", "value2");
    txMgr.commit();
    assertTrue(reg1.containsKey("key1"));
    assertTrue(reg1.containsValueForKey("key1"));
    assertEquals("key1", reg1.getEntry("key1").getKey());
    assertEquals("value1", reg1.getEntry("key1").getValue());
    assertEquals("value1", reg1.get("key1"));
    assertTrue(reg2.containsKey("key2"));
    assertTrue(reg2.containsValueForKey("key2"));
    assertEquals("key2", reg2.getEntry("key2").getKey());
    assertEquals("value2", reg2.getEntry("key2").getValue());
    assertEquals("value2", reg2.get("key2"));
    assertEquals(txCommitChanges + 2, stats.getTxCommitChanges());
    {
      List<EntryEvent<?, ?>> creates =
          TxEventTestUtil.getCreateEvents(te.getEvents());
      assertEquals(myTxId, te.getTransactionId());
      assertEquals(2, creates.size());

      for (EntryEvent ev : creates) {
        assertEquals(myTxId, ev.getTransactionId());
        assertTrue(ev.getRegion() == reg1 || ev.getRegion() == reg2);
        if (ev.getRegion() == reg1) {
          assertEquals("key1", ev.getKey());
          assertEquals("value1", ev.getNewValue());
        } else {
          assertEquals("key2", ev.getKey());
          assertEquals("value2", ev.getNewValue());
        }
        assertEquals(null, ev.getOldValue());
        verifyEventProps(ev);
        assertEquals(null, ev.getCallbackArgument());
        assertEquals(true, ev.isCallbackArgumentAvailable());
        assertTrue(!ev.isOriginRemote());
        assertTrue(!ev.getOperation().isExpiration());
        assertTrue(ev.getOperation().isDistributed());
      }
    }
    reg1.localDestroy("key1");
    reg2.localDestroy("key2");

    reg2.localDestroyRegion();
  }

  @Test
  public void testTxEvent() throws CacheException {
    TransactionId myTxId;
    Region<String, String> reg1 = region;

    txMgr.setListener(new TransactionListener() {
      @Override
      public void afterCommit(TransactionEvent event) {
        listenerAfterCommit = 1;
        te = event;
      }

      @Override
      public void afterFailedCommit(TransactionEvent event) {
        listenerAfterFailedCommit = 1;
        te = event;
      }

      @Override
      public void afterRollback(TransactionEvent event) {
        listenerAfterRollback = 1;
        te = event;
      }

      @Override
      public void close() {
        listenerClose = 1;
      }
    });

    // make sure each operation has the correct transaction event
    // check create
    txMgr.begin();
    myTxId = txMgr.getTransactionId();
    reg1.create("key1", "value1");
    txMgr.rollback();
    assertEquals(1, te.getEvents().size());
    assertEquals(0, TxEventTestUtil.getPutEvents(te.getEvents()).size());
    assertEquals(0, TxEventTestUtil.getInvalidateEvents(te.getEvents()).size());
    assertEquals(0, TxEventTestUtil.getDestroyEvents(te.getEvents()).size());
    {
      Cache teCache = te.getCache();
      assertEquals(teCache, cache);
      List<EntryEvent<?, ?>> creates =
          TxEventTestUtil.getCreateEvents(te.getEvents());
      assertEquals(myTxId, te.getTransactionId());
      assertEquals(1, creates.size());

      for (EntryEvent ev : creates) {
        assertEquals(myTxId, ev.getTransactionId());
        assertTrue(ev.getRegion() == reg1);
        assertEquals("key1", ev.getKey());
        assertEquals("value1", ev.getNewValue());
        assertEquals(null, ev.getOldValue());
        verifyEventProps(ev);
        assertEquals(null, ev.getCallbackArgument());
        assertEquals(true, ev.isCallbackArgumentAvailable());
        assertTrue(!ev.isOriginRemote());
        assertTrue(!ev.getOperation().isExpiration());
        assertTrue(ev.getOperation().isDistributed());
      }
    }

    // check put of existing entry
    reg1.create("key1", "value0");
    txMgr.begin();
    myTxId = txMgr.getTransactionId();
    reg1.put("key1", "value1");
    txMgr.rollback();
    assertEquals(1, te.getEvents().size());
    assertEquals(0, TxEventTestUtil.getCreateEvents(te.getEvents()).size());
    assertEquals(0, TxEventTestUtil.getInvalidateEvents(te.getEvents()).size());
    assertEquals(0, TxEventTestUtil.getDestroyEvents(te.getEvents()).size());
    {
      Cache teCache = te.getCache();
      assertEquals(teCache, cache);
      List<EntryEvent<?, ?>> creates = TxEventTestUtil.getPutEvents(te.getEvents());
      assertEquals(myTxId, te.getTransactionId());
      assertEquals(1, creates.size());

      for (EntryEvent ev : creates) {
        assertEquals(myTxId, ev.getTransactionId());
        assertTrue(ev.getRegion() == reg1);
        assertEquals("key1", ev.getKey());
        assertEquals("value1", ev.getNewValue());
        assertEquals("value0", ev.getOldValue());
        verifyEventProps(ev);
        assertEquals(null, ev.getCallbackArgument());
        assertEquals(true, ev.isCallbackArgumentAvailable());
        assertTrue(!ev.isOriginRemote());
        assertTrue(!ev.getOperation().isExpiration());
        assertTrue(ev.getOperation().isDistributed());
      }
    }
    reg1.localDestroy("key1");

    // check put of non-existent entry
    txMgr.begin();
    myTxId = txMgr.getTransactionId();
    reg1.put("key1", "value0");
    txMgr.rollback();
    assertEquals(1, te.getEvents().size());
    assertEquals(0, TxEventTestUtil.getPutEvents(te.getEvents()).size());
    assertEquals(0, TxEventTestUtil.getInvalidateEvents(te.getEvents()).size());
    assertEquals(0, TxEventTestUtil.getDestroyEvents(te.getEvents()).size());
    {
      Cache teCache = te.getCache();
      assertEquals(teCache, cache);
      List<EntryEvent<?, ?>> creates =
          TxEventTestUtil.getCreateEvents(te.getEvents());
      assertEquals(myTxId, te.getTransactionId());
      assertEquals(1, creates.size());

      for (EntryEvent ev : creates) {
        assertEquals(myTxId, ev.getTransactionId());
        assertTrue(ev.getRegion() == reg1);
        assertEquals("key1", ev.getKey());
        assertEquals("value0", ev.getNewValue());
        assertEquals(null, ev.getOldValue());
        verifyEventProps(ev);
        assertEquals(null, ev.getCallbackArgument());
        assertEquals(true, ev.isCallbackArgumentAvailable());
        assertTrue(!ev.isOriginRemote());
        assertTrue(!ev.getOperation().isExpiration());
        assertTrue(ev.getOperation().isDistributed());
      }
    }

    // check d invalidate of existing entry
    reg1.create("key1", "value0");
    txMgr.begin();
    myTxId = txMgr.getTransactionId();
    reg1.invalidate("key1");
    txMgr.rollback();
    assertEquals(1, te.getEvents().size());
    assertEquals(0, TxEventTestUtil.getCreateEvents(te.getEvents()).size());
    assertEquals(0, TxEventTestUtil.getPutEvents(te.getEvents()).size());
    assertEquals(0, TxEventTestUtil.getDestroyEvents(te.getEvents()).size());
    {
      Cache teCache = te.getCache();
      assertEquals(teCache, cache);
      List<EntryEvent<?, ?>> creates =
          TxEventTestUtil.getInvalidateEvents(te.getEvents());
      assertEquals(myTxId, te.getTransactionId());
      assertEquals(1, creates.size());

      for (EntryEvent ev : creates) {
        assertEquals(myTxId, ev.getTransactionId());
        assertTrue(ev.getRegion() == reg1);
        assertEquals("key1", ev.getKey());
        assertEquals(null, ev.getNewValue());
        assertEquals("value0", ev.getOldValue());
        verifyEventProps(ev);
        assertEquals(null, ev.getCallbackArgument());
        assertEquals(true, ev.isCallbackArgumentAvailable());
        assertTrue(!ev.isOriginRemote());
        assertTrue(!ev.getOperation().isExpiration());
        assertTrue(ev.getOperation().isDistributed());
      }
    }
    reg1.localDestroy("key1");

    // check l invalidate of existing entry
    reg1.create("key1", "value0");
    txMgr.begin();
    myTxId = txMgr.getTransactionId();
    reg1.localInvalidate("key1");
    txMgr.rollback();
    assertEquals(1, te.getEvents().size());
    assertEquals(0, TxEventTestUtil.getCreateEvents(te.getEvents()).size());
    assertEquals(0, TxEventTestUtil.getPutEvents(te.getEvents()).size());
    assertEquals(0, TxEventTestUtil.getDestroyEvents(te.getEvents()).size());
    {
      Cache teCache = te.getCache();
      assertEquals(teCache, cache);
      List<EntryEvent<?, ?>> creates =
          TxEventTestUtil.getInvalidateEvents(te.getEvents());
      assertEquals(myTxId, te.getTransactionId());
      assertEquals(1, creates.size());

      for (EntryEvent ev : creates) {
        assertEquals(myTxId, ev.getTransactionId());
        assertTrue(ev.getRegion() == reg1);
        assertEquals("key1", ev.getKey());
        assertEquals(null, ev.getNewValue());
        assertEquals("value0", ev.getOldValue());
        verifyEventProps(ev);
        assertEquals(null, ev.getCallbackArgument());
        assertEquals(true, ev.isCallbackArgumentAvailable());
        assertTrue(!ev.isOriginRemote());
        assertTrue(!ev.getOperation().isExpiration());
        if (!isPR()) {
          assertTrue(!ev.getOperation().isDistributed());
        }
      }
    }
    reg1.localDestroy("key1");

    // check d destroy of existing entry
    reg1.create("key1", "value0");
    txMgr.begin();
    myTxId = txMgr.getTransactionId();
    reg1.destroy("key1");
    txMgr.rollback();
    assertEquals(1, te.getEvents().size());
    assertEquals(0, TxEventTestUtil.getCreateEvents(te.getEvents()).size());
    assertEquals(0, TxEventTestUtil.getPutEvents(te.getEvents()).size());
    assertEquals(0, TxEventTestUtil.getInvalidateEvents(te.getEvents()).size());
    {
      Cache teCache = te.getCache();
      assertEquals(teCache, cache);
      List<EntryEvent<?, ?>> creates =
          TxEventTestUtil.getDestroyEvents(te.getEvents());
      assertEquals(myTxId, te.getTransactionId());
      assertEquals(1, creates.size());

      for (EntryEvent ev : creates) {
        assertEquals(myTxId, ev.getTransactionId());
        assertTrue(ev.getRegion() == reg1);
        assertEquals("key1", ev.getKey());
        assertEquals(null, ev.getNewValue());
        assertEquals("value0", ev.getOldValue());
        verifyEventProps(ev);
        assertEquals(null, ev.getCallbackArgument());
        assertEquals(true, ev.isCallbackArgumentAvailable());
        assertTrue(!ev.isOriginRemote());
        assertTrue(!ev.getOperation().isExpiration());
        assertTrue(ev.getOperation().isDistributed());
      }
    }
    reg1.localDestroy("key1");

    // check l destroy of existing entry
    reg1.create("key1", "value0");
    txMgr.begin();
    myTxId = txMgr.getTransactionId();
    reg1.localDestroy("key1");
    txMgr.rollback();
    assertEquals(1, te.getEvents().size());
    assertEquals(0, TxEventTestUtil.getCreateEvents(te.getEvents()).size());
    assertEquals(0, TxEventTestUtil.getPutEvents(te.getEvents()).size());
    assertEquals(0, TxEventTestUtil.getInvalidateEvents(te.getEvents()).size());
    {
      Cache teCache = te.getCache();
      assertEquals(teCache, cache);
      List<EntryEvent<?, ?>> creates =
          TxEventTestUtil.getDestroyEvents(te.getEvents());
      assertEquals(myTxId, te.getTransactionId());
      assertEquals(1, creates.size());

      for (EntryEvent ev : creates) {
        assertEquals(myTxId, ev.getTransactionId());
        assertTrue(ev.getRegion() == reg1);
        assertEquals("key1", ev.getKey());
        assertEquals(null, ev.getNewValue());
        assertEquals("value0", ev.getOldValue());
        verifyEventProps(ev);
        assertEquals(null, ev.getCallbackArgument());
        assertEquals(true, ev.isCallbackArgumentAvailable());
        assertTrue(!ev.isOriginRemote());
        assertTrue(!ev.getOperation().isExpiration());
        if (!isPR()) {
          assertTrue(!ev.getOperation().isDistributed());
        }
      }
    }
    reg1.localDestroy("key1");
  }

  private static class CountingCallBackValidator {
    List<Object> asserts;
    final String createWriterAssert = "create writer Assert";
    final String createListenerAssert = "create listener Assert";
    final String updateWriterAssert = "update writer Assert";
    final String updateListenerAssert = "update listener Assert";
    final String invalAssert = "invalidate Assert";
    final String destroyWriterAssert = "destroy writer Assert";
    final String destroyListenerAssert = "destroy listener Assert";
    final String localDestroyWriterAssert = "local destroy writer Assert";
    final String localDestroyListenerAssert = "local destroy listener Assert";

    CountingCacheListener cl;
    CountingCacheWriter cw;

    CountingCallBackValidator(CountingCacheListener cl, CountingCacheWriter cw) {
      this.cl = cl;
      this.cw = cw;
      asserts = new ArrayList<>(8);
    }

    void assertCreateWriterCnt(int cnt) {
      assertCreateWriterCnt(cnt, true);
    }

    void assertCreateWriterCnt(int cnt, boolean remember) {
      if (remember) {
        asserts.add(createWriterAssert);
        asserts.add(cnt);
      }
      assertEquals(cnt, cw.getBeforeCreateCalls());
    }

    void assertCreateListenerCnt(int cnt) {
      assertCreateListenerCnt(cnt, true);
    }

    void assertCreateListenerCnt(int cnt, boolean remember) {
      if (remember) {
        asserts.add(createListenerAssert);
        asserts.add(cnt);
      }
      assertEquals(cnt, cl.getAfterCreateCalls());
    }

    void assertDestroyWriterCnt(int cnt) {
      assertDestroyWriterCnt(cnt, true);
    }

    void assertDestroyWriterCnt(int cnt, boolean remember) {
      if (remember) {
        asserts.add(destroyWriterAssert);
        asserts.add(cnt);
      }
      assertEquals(cnt, cw.getBeforeDestroyCalls(false));
    }

    void assertDestroyListenerCnt(int cnt) {
      assertDestroyListenerCnt(cnt, true);
    }

    void assertDestroyListenerCnt(int cnt, boolean remember) {
      if (remember) {
        asserts.add(destroyListenerAssert);
        asserts.add(cnt);
      }
      assertEquals(cnt, cl.getAfterDestroyCalls(false));
    }

    void assertLocalDestroyWriterCnt(int cnt) {
      assertLocalDestroyWriterCnt(cnt, true);
    }

    void assertLocalDestroyWriterCnt(int cnt, boolean remember) {
      if (remember) {
        asserts.add(localDestroyWriterAssert);
        asserts.add(cnt);
      }
      assertEquals(0, cw.getBeforeDestroyCalls(true));
    }

    void assertLocalDestroyListenerCnt(int cnt) {
      assertLocalDestroyListenerCnt(cnt, true);
    }

    void assertLocalDestroyListenerCnt(int cnt, boolean remember) {
      if (remember) {
        asserts.add(localDestroyListenerAssert);
        asserts.add(cnt);
      }
      assertEquals(cnt, cl.getAfterDestroyCalls(true));
    }

    void assertUpdateWriterCnt(int cnt) {
      assertUpdateWriterCnt(cnt, true);
    }

    void assertUpdateWriterCnt(int cnt, boolean remember) {
      if (remember) {
        asserts.add(updateWriterAssert);
        asserts.add(cnt);
      }
      assertEquals(cnt, cw.getBeforeUpdateCalls());
    }

    void assertUpdateListenerCnt(int cnt) {
      assertUpdateListenerCnt(cnt, true);
    }

    void assertUpdateListenerCnt(int cnt, boolean remember) {
      if (remember) {
        asserts.add(updateListenerAssert);
        asserts.add(cnt);
      }
      assertEquals(cnt, cl.getAfterUpdateCalls());
    }

    void assertInvalidateCnt(int cnt) {
      assertInvalidateCnt(cnt, true);
    }

    void assertInvalidateCnt(int cnt, boolean remember) {
      if (remember) {
        asserts.add(invalAssert);
        asserts.add(cnt);
      }
      assertEquals(cnt, cl.getAfterInvalidateCalls());
    }

    void reAssert() {
      Iterator assertItr = asserts.iterator();
      String assertType;
      Integer count;
      int cnt;
      while (assertItr.hasNext()) {
        assertType = (String) assertItr.next();
        assertTrue("CountingCallBackValidator reassert, did not have an associated count",
            assertItr.hasNext());
        count = (Integer) assertItr.next();
