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

import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.geode.distributed.ConfigurationProperties.LOG_LEVEL;
import static org.apache.geode.distributed.ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER;
import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort;
import static org.apache.geode.internal.cache.ExpiryTask.permitExpiration;
import static org.apache.geode.internal.cache.ExpiryTask.suspendExpiration;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

import java.io.IOException;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CountDownLatch;

import javax.naming.Context;
import javax.naming.NamingException;

import jakarta.transaction.RollbackException;
import jakarta.transaction.Status;
import jakarta.transaction.UserTransaction;
import org.awaitility.core.ThrowingRunnable;
import org.junit.Ignore;
import org.junit.Test;

import org.apache.geode.ExpirationDetector;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.AttributesMutator;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.CacheEvent;
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.EntryEvent;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.ExpirationAction;
import org.apache.geode.cache.ExpirationAttributes;
import org.apache.geode.cache.InterestPolicy;
import org.apache.geode.cache.InterestResultPolicy;
import org.apache.geode.cache.LoaderHelper;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.Region.Entry;
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.SubscriptionAttributes;
import org.apache.geode.cache.TransactionDataNotColocatedException;
import org.apache.geode.cache.TransactionDataRebalancedException;
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.TransactionWriter;
import org.apache.geode.cache.TransactionWriterException;
import org.apache.geode.cache.UnsupportedOperationInTransactionException;
import org.apache.geode.cache.client.ClientCache;
import org.apache.geode.cache.client.ClientCacheFactory;
import org.apache.geode.cache.client.ClientRegionFactory;
import org.apache.geode.cache.client.ClientRegionShortcut;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionAdapter;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.execute.RegionFunctionContext;
import org.apache.geode.cache.partition.PartitionRegionHelper;
import org.apache.geode.cache.query.CqEvent;
import org.apache.geode.cache.query.CqListener;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.cache.util.CacheWriterAdapter;
import org.apache.geode.distributed.ConfigurationProperties;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.execute.CustomerIDPartitionResolver;
import org.apache.geode.internal.cache.execute.InternalFunctionService;
import org.apache.geode.internal.cache.execute.data.CustId;
import org.apache.geode.internal.cache.execute.data.Customer;
import org.apache.geode.internal.cache.execute.data.Order;
import org.apache.geode.internal.cache.execute.data.OrderId;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.test.awaitility.GeodeAwaitility;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.SerializableCallable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;


public class RemoteTransactionDUnitTest extends JUnit4CacheTestCase {

  protected final String CUSTOMER = "custRegion";
  protected final String ORDER = "orderRegion";
  protected final String D_REFERENCE = "distrReference";

  private final SerializableCallable<Object> getNumberOfTXInProgress =
      new SerializableCallable<Object>() {
        @Override
        public Object call() {
          TXManagerImpl mgr = getCache().getTxManager();
          return mgr.hostedTransactionsInProgressForTest();
        }
      };
  private final SerializableCallable<Object> verifyNoTxState = new SerializableCallable<Object>() {
    @Override
    public Object call() {
      try {
        CacheFactory.getAnyInstance();
      } catch (CacheClosedException e) {
        return null;
      }
      // TXManagerImpl mgr = getCache().getTxManager();
      // assertIndexDetailsEquals(0, mgr.hostedTransactionsInProgressForTest());
      final TXManagerImpl mgr = getCache().getTxManager();
      GeodeAwaitility.await().atMost(Duration.ofMillis(30000)).untilAsserted(
          () -> assertThat(mgr.hostedTransactionsInProgressForTest()).isZero());
      return null;
    }
  };

  protected enum OP {
    PUT, GET, DESTROY, INVALIDATE, KEYS, VALUES, ENTRIES, PUTALL, GETALL, REMOVEALL
  }

  @Override
  public Properties getDistributedSystemProperties() {
    Properties result = super.getDistributedSystemProperties();
    result.put(ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER,
        "org.apache.geode.internal.cache.RemoteTransactionDUnitTest*"
            + ";org.apache.geode.test.dunit.**" + ";org.apache.geode.test.junit.**"
            + ";org.apache.geode.internal.cache.execute.data.CustId"
            + ";org.apache.geode.internal.cache.execute.data.Customer");
    return result;
  }

  @Override
  public final void preTearDownCacheTestCase() {
    try {
      Invoke.invokeInEveryVM(verifyNoTxState);
    } finally {
      closeAllCache();
    }
  }

  void createRegion(boolean accessor, int redundantCopies, InterestPolicy interestPolicy) {
    AttributesFactory<Object, Object> af = new AttributesFactory<Object, Object>();
    af.setScope(Scope.DISTRIBUTED_ACK);
    af.setDataPolicy(DataPolicy.REPLICATE);
    af.setConcurrencyChecksEnabled(getConcurrencyChecksEnabled());
    getCache().createRegion(D_REFERENCE, af.create());
    af = new AttributesFactory<Object, Object>();
    af.setConcurrencyChecksEnabled(getConcurrencyChecksEnabled());
    if (interestPolicy != null) {
      af.setSubscriptionAttributes(new SubscriptionAttributes(interestPolicy));
    }
    af.setPartitionAttributes(new PartitionAttributesFactory<>()
        .setTotalNumBuckets(4).setLocalMaxMemory(accessor ? 0 : 1)
        .setPartitionResolver(new CustomerIDPartitionResolver("resolver1"))
        .setRedundantCopies(redundantCopies).create());
    getCache().createRegion(CUSTOMER, af.create());
    af.setPartitionAttributes(new PartitionAttributesFactory<>().setTotalNumBuckets(4)
        .setLocalMaxMemory(accessor ? 0 : 1)
        .setPartitionResolver(new CustomerIDPartitionResolver("resolver2"))
        .setRedundantCopies(redundantCopies).setColocatedWith(CUSTOMER).create());
    getCache().createRegion(ORDER, af.create());
  }

  protected boolean getConcurrencyChecksEnabled() {
    return false;
  }

  void populateData() {
    Region<Object, Object> custRegion = getCache().getRegion(CUSTOMER);
    Region<Object, Object> orderRegion = getCache().getRegion(ORDER);
    Region<Object, Object> refRegion = getCache().getRegion(D_REFERENCE);
    for (int i = 0; i < 5; i++) {
      CustId custId = new CustId(i);
      Customer customer = new Customer("customer" + i, "address" + i);
      OrderId orderId = new OrderId(i, custId);
      Order order = new Order("order" + i);
      custRegion.put(custId, customer);
      orderRegion.put(orderId, order);
      refRegion.put(custId, customer);
    }
  }

  protected void initAccessorAndDataStore(VM accessor, VM datastore, final int redundantCopies) {
    accessor.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        createRegion(true/* accessor */, redundantCopies, null);
        return null;
      }
    });

    datastore.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        createRegion(false/* accessor */, redundantCopies, null);
        populateData();
        return null;
      }
    });
  }

  protected void initAccessorAndDataStore(VM accessor, VM datastore1, VM datastore2,
      final int redundantCopies) {
    datastore2.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        createRegion(false/* accessor */, redundantCopies, null);
        return null;
      }
    });

    initAccessorAndDataStore(accessor, datastore1, redundantCopies);
  }

  private void initAccessorAndDataStoreWithInterestPolicy(VM accessor, VM datastore1, VM datastore2,
      final int redundantCopies) {
    datastore2.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        createRegion(false/* accessor */, redundantCopies, InterestPolicy.ALL);
        return null;
      }
    });

    initAccessorAndDataStore(accessor, datastore1, redundantCopies);
  }

  protected class DoOpsInTX extends SerializableCallable<Object> {
    private final OP op;
    Customer expectedCust;
    Customer expectedRefCust = null;
    Order expectedOrder;
    Order expectedOrder2;
    Order expectedOrder3;

    DoOpsInTX(OP op) {
      this.op = op;
    }

    @Override
    public Object call() {
      CacheTransactionManager mgr = getCache().getTxManager();
      LogWriterUtils.getLogWriter().fine("testTXPut starting tx");
      mgr.begin();
      Region<CustId, Customer> custRegion = getCache().getRegion(CUSTOMER);
      Region<OrderId, Order> orderRegion = getCache().getRegion(ORDER);
      Region<CustId, Customer> refRegion = getCache().getRegion(D_REFERENCE);
      CustId custId = new CustId(1);
      OrderId orderId = new OrderId(1, custId);
      OrderId orderId2 = new OrderId(2, custId);
      OrderId orderId3 = new OrderId(3, custId);
      switch (op) {
        case PUT:
          expectedCust = new Customer("foo", "bar");
          expectedOrder = new Order("fooOrder");
          expectedOrder2 = new Order("fooOrder2");
          expectedOrder3 = new Order("fooOrder3");
          custRegion.put(custId, expectedCust);
          orderRegion.put(orderId, expectedOrder);
          Map<OrderId, Order> orders = new HashMap<OrderId, Order>();
          orders.put(orderId2, expectedOrder2);
          orders.put(orderId3, expectedOrder3);
          getCache().getLogger().fine("SWAP:doingPutAll");
          refRegion.put(custId, expectedCust);
          Set<OrderId> ordersSet = new HashSet<>();
          ordersSet.add(orderId);
          ordersSet.add(orderId2);
          ordersSet.add(orderId3);
          break;
        case GET:
          expectedCust = custRegion.get(custId);
          expectedOrder = orderRegion.get(orderId);
          expectedRefCust = refRegion.get(custId);
          assertThat(expectedCust).isNotNull();
          assertThat(expectedOrder).isNotNull();
          assertThat(expectedRefCust).isNotNull();
          validateContains(custId, Collections.singleton(orderId), true, true);
          break;
        case DESTROY:
          validateContains(custId, Collections.singleton(orderId), true);
          custRegion.destroy(custId);
          orderRegion.destroy(orderId);
          refRegion.destroy(custId);
          validateContains(custId, Collections.singleton(orderId), false);
          break;
        case REMOVEALL:
          validateContains(custId, Collections.singleton(orderId), true);
          custRegion.removeAll(Collections.singleton(custId));
          orderRegion.removeAll(Collections.singleton(orderId));
          refRegion.removeAll(Collections.singleton(custId));
          validateContains(custId, Collections.singleton(orderId), false);
          break;
        case INVALIDATE:
          validateContains(custId, Collections.singleton(orderId), true);
          custRegion.invalidate(custId);
          orderRegion.invalidate(orderId);
          refRegion.invalidate(custId);
          validateContains(custId, Collections.singleton(orderId), true, false);
          break;
        default:
          throw new IllegalStateException();
      }
      return mgr.suspend();
    }
  }

  void validateContains(CustId custId, Set<OrderId> orderId, boolean doesIt) {
    validateContains(custId, orderId, doesIt, doesIt);
  }

  void validateContains(CustId custId, Set<OrderId> ordersSet, boolean containsKey,
      boolean containsValue) {
    Region<CustId, Customer> custRegion = getCache().getRegion(CUSTOMER);
    Region<OrderId, Order> orderRegion = getCache().getRegion(ORDER);
    Region<CustId, Order> refRegion = getCache().getRegion(D_REFERENCE);
    boolean rContainsKC = custRegion.containsKey(custId);
    boolean rContainsKO = containsKey;
    for (OrderId o : ordersSet) {
      getCache().getLogger()
          .fine("SWAP:rContainsKO:" + rContainsKO + " containsKey:" + orderRegion.containsKey(o));
      rContainsKO = rContainsKO && orderRegion.containsKey(o);
    }
    boolean rContainsKR = refRegion.containsKey(custId);

    boolean rContainsVC = custRegion.containsValueForKey(custId);
    boolean rContainsVO = containsValue;
    for (OrderId o : ordersSet) {
      rContainsVO = rContainsVO && orderRegion.containsValueForKey(o);
    }
    boolean rContainsVR = refRegion.containsValueForKey(custId);

    assertThat(containsKey).isEqualTo(rContainsKC);
    assertThat(containsKey).isEqualTo(rContainsKO);
    assertThat(containsKey).isEqualTo(rContainsKR);
    assertThat(containsValue).isEqualTo(rContainsVR);
    assertThat(containsValue).isEqualTo(rContainsVC);
    assertThat(containsValue).isEqualTo(rContainsVO);

    if (containsKey) {
      Region.Entry eC = custRegion.getEntry(custId);
      for (OrderId o : ordersSet) {
        assertThat(orderRegion.getEntry(o)).isNotNull();
      }
      Region.Entry eR = refRegion.getEntry(custId);
      assertThat(eC).isNotNull();
      assertThat(eR).isNotNull();
      // assertIndexDetailsEquals(1,custRegion.size());
      // assertIndexDetailsEquals(1,orderRegion.size());
      // assertIndexDetailsEquals(1,refRegion.size());

    } else {
      // assertIndexDetailsEquals(0,custRegion.size());
      // assertIndexDetailsEquals(0,orderRegion.size());
      // assertIndexDetailsEquals(0,refRegion.size());
      try {
        Region.Entry eC = custRegion.getEntry(custId);
        assertThat(eC).as("should have had an EntryNotFoundException:" + eC).isNull();

      } catch (EntryNotFoundException enfe) {
        // this is what we expect
      }
      try {
        for (OrderId o : ordersSet) {
          assertThat(orderRegion.getEntry(o))
              .as("should have had an EntryNotFoundException:" + orderRegion.getEntry(o))
              .isNull();
        }
      } catch (EntryNotFoundException enfe) {
        // this is what we expect
      }
      try {
        Region.Entry eR = refRegion.getEntry(custId);
        assertThat(eR).as("should have had an EntryNotFoundException:" + eR).isNull();
      } catch (EntryNotFoundException enfe) {
        // this is what we expect
      }

    }

  }


  void verifyAfterCommit(OP op) {
    Region<CustId, Customer> custRegion = getCache().getRegion(CUSTOMER);
    Region<CustId, Customer> refRegion = getCache().getRegion(D_REFERENCE);
    Region<OrderId, Order> orderRegion = getCache().getRegion(ORDER);
    CustId custId = new CustId(1);
    OrderId orderId = new OrderId(1, custId);
    OrderId orderId2 = new OrderId(2, custId);
    OrderId orderId3 = new OrderId(3, custId);
    Customer expectedCust;
    Order expectedOrder;
    Order expectedOrder2;
    Order expectedOrder3;
    Customer expectedRef;
    switch (op) {
      case PUT:
        expectedCust = new Customer("foo", "bar");
        expectedOrder = new Order("fooOrder");
        expectedOrder2 = new Order("fooOrder2");
        expectedOrder3 = new Order("fooOrder3");
        expectedRef = expectedCust;
        assertThat(custRegion.getEntry(custId)).isNotNull();
        assertThat(expectedCust).isEqualTo(custRegion.getEntry(custId).getValue());
        /*
         * assertThat(orderRegion.getEntry(orderId)).isNotNull();
         * assertIndexDetailsEquals(expectedOrder,
         * orderRegion.getEntry(orderId).getValue());
         *
         * assertThat(orderRegion.getEntry(orderId2)).isNotNull();
         * assertIndexDetailsEquals(expectedOrder2,
         * orderRegion.getEntry(orderId2).getValue());
         *
         * assertThat(orderRegion.getEntry(orderId3)).isNotNull();
         * assertIndexDetailsEquals(expectedOrder3,
         * orderRegion.getEntry(orderId3).getValue());
         */
        assertThat(refRegion.getEntry(custId)).isNotNull();
        assertThat(expectedRef).isEqualTo(refRegion.getEntry(custId).getValue());

        // Set<OrderId> ordersSet = new HashSet<>();
        // ordersSet.add(orderId);ordersSet.add(orderId2);ordersSet.add(orderId3);
        // validateContains(custId, ordersSet, true);
        break;
      case GET:
        expectedCust = custRegion.get(custId);
        expectedOrder = orderRegion.get(orderId);
        expectedRef = refRegion.get(custId);
        validateContains(custId, Collections.singleton(orderId), true);
        break;
      case DESTROY:
        assertThat(!custRegion.containsKey(custId)).isTrue();
        assertThat(!orderRegion.containsKey(orderId)).isTrue();
        assertThat(!refRegion.containsKey(custId)).isTrue();
        validateContains(custId, Collections.singleton(orderId), false);
        break;
      case INVALIDATE:
        boolean validateContainsKey = true;
        if (!((GemFireCacheImpl) custRegion.getCache()).isClient()) {
          assertThat(custRegion.containsKey(custId)).isTrue();
          assertThat(orderRegion.containsKey(orderId)).isTrue();
          assertThat(refRegion.containsKey(custId)).isTrue();
        }
        assertThat(custRegion.get(custId)).isNull();
        assertThat(orderRegion.get(orderId)).isNull();
        assertThat(refRegion.get(custId)).isNull();
        validateContains(custId, Collections.singleton(orderId), validateContainsKey, false);
        break;
      default:
        throw new IllegalStateException();
    }
  }


  void verifyAfterRollback(OP op) {
    Region<CustId, Customer> custRegion = getCache().getRegion(CUSTOMER);
    Region<OrderId, Order> orderRegion = getCache().getRegion(ORDER);
    Region<CustId, Customer> refRegion = getCache().getRegion(D_REFERENCE);
    assertThat(custRegion).isNotNull();
    assertThat(orderRegion).isNotNull();
    assertThat(refRegion).isNotNull();

    CustId custId = new CustId(1);
    OrderId orderId = new OrderId(1, custId);
    OrderId orderId2 = new OrderId(2, custId);
    OrderId orderId3 = new OrderId(3, custId);
    Customer expectedCust;
    Order expectedOrder;
    Customer expectedRef;
    switch (op) {
      case PUT:
        expectedCust = new Customer("customer1", "address1");
        expectedOrder = new Order("order1");
        expectedRef = new Customer("customer1", "address1");
        assertThat(expectedCust).isEqualTo(custRegion.getEntry(custId).getValue());
        assertThat(expectedOrder).isEqualTo(orderRegion.getEntry(orderId).getValue());
        getCache().getLogger().info("SWAP:verifyRollback:" + orderRegion);
        getCache().getLogger().info("SWAP:verifyRollback:" + orderRegion.getEntry(orderId2));
        assertThat(getCache().getTXMgr().getTXState()).isNull();
        assertThat(orderRegion.getEntry(orderId2))
            .as("" + orderRegion.getEntry(orderId2)).isNull();
        assertThat(orderRegion.getEntry(orderId3)).isNull();
        assertThat(orderRegion.get(orderId2)).isNull();
        assertThat(orderRegion.get(orderId3)).isNull();
        assertThat(expectedRef).isEqualTo(refRegion.getEntry(custId).getValue());
        validateContains(custId, Collections.singleton(orderId), true);
        break;
      case GET:
        expectedCust = custRegion.getEntry(custId).getValue();
        expectedOrder = orderRegion.getEntry(orderId).getValue();
        expectedRef = refRegion.getEntry(custId).getValue();
        validateContains(custId, Collections.singleton(orderId), true);
        break;
      case DESTROY:
        assertThat(custRegion.containsKey(custId)).isFalse();
        assertThat(orderRegion.containsKey(orderId)).isFalse();
        assertThat(refRegion.containsKey(custId)).isFalse();
        validateContains(custId, Collections.singleton(orderId), true);
        break;
      case INVALIDATE:
        assertThat(custRegion.containsKey(custId)).isTrue();
        assertThat(orderRegion.containsKey(orderId)).isTrue();
        assertThat(refRegion.containsKey(custId)).isTrue();
        assertThat(custRegion.get(custId)).isNull();
        assertThat(orderRegion.get(orderId)).isNull();
        assertThat(refRegion.get(custId)).isNull();
        validateContains(custId, Collections.singleton(orderId), true, true);
        break;
      default:
        throw new IllegalStateException();
    }
  }


  @Test
  public void testTXCreationAndCleanupAtCommit() {
    doBasicChecks(true);
  }

  @Test
  public void testTXCreationAndCleanupAtRollback() {
    doBasicChecks(false);
  }

  private void doBasicChecks(final boolean commit) {
    Host host = Host.getHost(0);
    VM accessor = host.getVM(0);
    VM datastore = host.getVM(1);
    initAccessorAndDataStore(accessor, datastore, 0);

    final TXId txId = (TXId) accessor.invoke(new DoOpsInTX(OP.PUT));

    datastore.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        TXManagerImpl mgr = getCache().getTxManager();
        assertThat(mgr).isNotNull();
        assertThat(mgr.isHostedTxInProgress(txId)).isTrue();
        return null;
      }
    });

    accessor.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        TXManagerImpl mgr = getCache().getTxManager();
        mgr.resume(txId);
        TXStateProxy tx = mgr.pauseTransaction();
        assertThat(tx).isNotNull();
        mgr.unpauseTransaction(tx);
        if (commit) {
          mgr.commit();
        } else {
          mgr.rollback();
        }
        return null;
      }
    });

    datastore.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        TXManagerImpl mgr = getCache().getTxManager();
        assertThat(mgr.isHostedTxInProgress(txId)).isFalse();
        return null;
      }
    });
    if (commit) {
      accessor.invoke(new SerializableCallable<Object>() {
        @Override
        public Object call() {
          verifyAfterCommit(OP.PUT);
          return null;
        }
      });
    } else {
      accessor.invoke(new SerializableCallable<Object>() {
        @Override
        public Object call() {
          verifyAfterRollback(OP.PUT);
          return null;
        }
      });
    }
  }

  @Test
  public void testPRTXGet() {
    Host host = Host.getHost(0);
    VM accessor = host.getVM(0);
    VM datastore = host.getVM(1);
    initAccessorAndDataStore(accessor, datastore, 0);

    final TXId txId = (TXId) accessor.invoke(new DoOpsInTX(OP.GET));

    datastore.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        TXManagerImpl mgr = getCache().getTxManager();
        assertThat(mgr.isHostedTxInProgress(txId)).isTrue();
        TXStateProxy tx = mgr.getHostedTXState(txId);
        System.out.println("TXRS:" + tx.getRegions());
        assertThat(tx.getRegions().size()).isEqualTo(3);// 2 buckets for the two puts we
        // did in the accessor
        // plus the dist. region
        for (InternalRegion r : tx.getRegions()) {
          assertThat(r instanceof DistributedRegion).isTrue();
          TXRegionState rs = tx.readRegion(r);
          for (Object key : rs.getEntryKeys()) {
            TXEntryState es = rs.readEntry(key);
            assertThat(es.getValue(key, r, false)).isNotNull();
            assertThat(es.isDirty()).isFalse();
          }
        }
        return null;
      }
    });

    accessor.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        CacheTransactionManager mgr = getCache().getTxManager();
        mgr.resume(txId);
        mgr.commit();
        verifyAfterCommit(OP.GET);
        return null;
      }
    });
  }

  @Test
  public void testPRTXGetOnRemoteWithLoader() {
    Host host = Host.getHost(0);
    VM accessor = host.getVM(0);
    VM datastore = host.getVM(1);
    initAccessorAndDataStore(accessor, datastore, 0);

    datastore.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        AttributesMutator<Object, Object> am =
            getCache().getRegion(CUSTOMER).getAttributesMutator();
        am.setCacheLoader(new CacheLoader() {
          @Override
          public Object load(LoaderHelper helper) throws CacheLoaderException {
            return new Customer("sup dawg", "add");
          }

          @Override
          public void close() {}
        });
        return null;
      }
    });

    accessor.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        TXManagerImpl mgr = getCache().getTxManager();
        mgr.begin();
        Region<Object, Object> cust = getCache().getRegion(CUSTOMER);
        Customer s = (Customer) cust.get(new CustId(8));
        assertThat(new Customer("sup dawg", "add")).isEqualTo(s);
        assertThat(cust.containsKey(new CustId(8))).isTrue();
        TXStateProxy tx = mgr.pauseTransaction();
        assertThat(cust.containsKey(new CustId(8))).isFalse();
        mgr.unpauseTransaction(tx);
        mgr.commit();
        Customer s2 = (Customer) cust.get(new CustId(8));
        Customer ex = new Customer("sup dawg", "add");
        assertThat(ex).isEqualTo(s);
        assertThat(ex).isEqualTo(s2);
        return null;
      }
    });
  }

  /**
   * Make sure that getEntry returns null properly and values when it should
   */
  @Test
  public void testPRTXGetEntryOnRemoteSide() {
    Host host = Host.getHost(0);
    VM accessor = host.getVM(0);
    VM datastore = host.getVM(1);
    initAccessorAndDataStore(accessor, datastore, 0);

    datastore.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        return null;
      }
    });

    accessor.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        Region<Object, Object> cust = getCache().getRegion(CUSTOMER);
        CustId sup = new CustId(7);
        Region.Entry e = cust.getEntry(sup);
        assertThat(e).isNull();
        CustId custId = new CustId(5);
        cust.put(custId, new Customer("customer5", "address5"));

        Region.Entry ee = cust.getEntry(custId);
        assertThat(ee).isNotNull();

        CacheTransactionManager mgr = getCache().getTxManager();
        mgr.begin();
        Region.Entry e2 = cust.getEntry(sup);
        assertThat(e2).isNull();
        mgr.commit();
        Region.Entry e3 = cust.getEntry(sup);
        assertThat(e3).isNull();

        mgr.begin();
        Customer dawg = new Customer("dawg", "dawgaddr");
        cust.put(sup, dawg);
        Region.Entry e4 = cust.getEntry(sup);
        assertThat(e4).isNotNull();
        assertThat(dawg).isEqualTo(e4.getValue());
        mgr.commit();

        Region.Entry e5 = cust.getEntry(sup);
        assertThat(e5).isNotNull();
        assertThat(dawg).isEqualTo(e5.getValue());
        return null;
      }
    });
  }


  @Test
  public void testPRTXGetOnLocalWithLoader() {
    Host host = Host.getHost(0);
    VM accessor = host.getVM(0);
    VM datastore = host.getVM(1);
    initAccessorAndDataStore(accessor, datastore, 0);

    datastore.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        AttributesMutator<Object, Object> am =
            getCache().getRegion(CUSTOMER).getAttributesMutator();
        am.setCacheLoader(new CacheLoader<Object, Object>() {
          @Override
          public Object load(LoaderHelper<Object, Object> helper) throws CacheLoaderException {
            return new Customer("sup dawg", "addr");
          }

          @Override
          public void close() {}
        });
        CacheTransactionManager mgr = getCache().getTxManager();
        mgr.begin();
        Region cust = getCache().getRegion(CUSTOMER);
        CustId custId = new CustId(6);
        Customer s = (Customer) cust.get(custId);
        mgr.commit();
        Customer s2 = (Customer) cust.get(custId);
        Customer expectedCust = new Customer("sup dawg", "addr");
        assertThat(s).isEqualTo(expectedCust);
        assertThat(s2).isEqualTo(expectedCust);
        return null;
      }
    });

    accessor.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        return null;
      }
    });
  }


  @Test
  public void testTXPut() {
    Host host = Host.getHost(0);
    VM acc = host.getVM(0);
    VM datastore = host.getVM(1);
    initAccessorAndDataStore(acc, datastore, 0);
    VM accessor = getVMForTransactions(acc, datastore);

    final TXId txId = (TXId) accessor.invoke(new DoOpsInTX(OP.PUT));

    datastore.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        TXManagerImpl mgr = getCache().getTxManager();
        assertThat(mgr.isHostedTxInProgress(txId)).isTrue();
        TXStateProxy tx = mgr.getHostedTXState(txId);
        assertThat(tx.getRegions().size()).isEqualTo(3);// 2 buckets for the two puts we
        // did in the accessor
        // +1 for the dist_region
        for (InternalRegion r : tx.getRegions()) {
          assertThat(r instanceof DistributedRegion).isTrue();
          TXRegionState rs = tx.readRegion(r);
          for (Object key : rs.getEntryKeys()) {
            TXEntryState es = rs.readEntry(key);
            assertThat(es.getValue(key, r, false)).isNotNull();
            assertThat(es.isDirty()).isTrue();
          }
        }
        return null;
      }
    });
    accessor.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        TXManagerImpl mgr = getCache().getTxManager();
        mgr.resume(txId);
        assertThat(mgr.getTXState()).isNotNull();
        getCache().getLogger().fine("SWAP:accessorTXState:" + mgr.getTXState());
        mgr.commit();
        verifyAfterCommit(OP.PUT);
        assertThat(mgr.getTXState()).isNull();
        return null;
      }
    });
  }

  @Test
  public void testTXInvalidate() {
    Host host = Host.getHost(0);
    VM acc = host.getVM(0);
