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

import static org.assertj.core.api.Assertions.catchThrowable;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

import org.apache.logging.log4j.Logger;
import org.assertj.core.api.Assertions;
import org.junit.Ignore;
import org.junit.Test;
import util.TestException;

import org.apache.geode.cache.CacheTransactionManager;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.Region.Entry;
import org.apache.geode.cache.TransactionDataNotColocatedException;
import org.apache.geode.cache.TransactionDataRebalancedException;
import org.apache.geode.cache.TransactionException;
import org.apache.geode.cache.TransactionId;
import org.apache.geode.cache.execute.Execution;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionException;
import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.partition.PartitionRegionHelper;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.NanoTimer;
import org.apache.geode.internal.cache.ForceReattemptException;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.TXManagerImpl;
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.execute.data.Shipment;
import org.apache.geode.internal.cache.execute.data.ShipmentId;
import org.apache.geode.internal.cache.partitioned.PRLocallyDestroyedException;
import org.apache.geode.internal.cache.partitioned.PartitionedRegionRebalanceOp;
import org.apache.geode.internal.cache.partitioned.rebalance.BucketOperatorImpl;
import org.apache.geode.internal.cache.partitioned.rebalance.ExplicitMoveDirector;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.test.dunit.Assert;
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.SerializableRunnable;

/**
 * Test for co-located PR transactions. the test creates two data hosts and uses function execution
 * to execute transactions.
 */
public class PRTransactionDUnitTest extends PRColocationDUnitTest {
  private static final Logger logger = LogService.getLogger();

  static final int VERIFY_TX = 0;

  static final int VERIFY_ROLLBACK = 1;

  static final int VERIFY_NON_COLOCATION = 2;

  static final int VERIFY_DESTROY = 4;

  static final int VERIFY_LISTENER_CALLBACK = 5;

  static final int VERIFY_INVALIDATE = 6;

  static final int VERIFY_TXSTATE_CONFLICT = 7;

  static final int VERIFY_REP_READ = 8;

  static final int UPDATE_NON_COLOCATION = 9;

  private final int totalIterations = 50;

  private final int warmupIterations = 10;

  private final int perfOrderShipmentPairs = 1;

  public PRTransactionDUnitTest() {
    super();
  }

  @Test
  public void testBasicPRTransactionRedundancy0() {
    basicPRTXInFunction(0, true);
  }

  @Test
  public void testBasicPRTransactionRedundancy1() {
    basicPRTXInFunction(1, true);
  }

  @Test
  public void testBasicPRTransactionRedundancy2() {
    basicPRTXInFunction(2, true);
  }

  @Test
  public void testBasicPRTransactionNoDataRedundancy0() {
    basicPRTXInFunction(0, false);
  }

  @Test
  public void testBasicPRTransactionNoDataRedundancy1() {
    basicPRTXInFunction(1, false);
  }

  @Test
  public void testBasicPRTransactionNoDataRedundancy2() {
    basicPRTXInFunction(2, false);
  }

  @Test
  public void testBasicPRTransactionNonColocatedFunction0() {
    basicPRTXInNonColocatedFunction(0);
  }

  /**
   * Test two non colocated functions in a transaction. This method invokes
   * {@link MyTransactionFunction} and tells it what to test, using different arguments.
   *
   * @param redundantBuckets redundant buckets for colocated PRs
   */
  private void basicPRTXInNonColocatedFunction(int redundantBuckets) {
    setupColocatedRegions(redundantBuckets);

    dataStore1.invoke(() -> registerFunction());
    dataStore2.invoke(() -> registerFunction());

    dataStore1.invoke(() -> runTXFunctions());
  }

  private void registerFunction() {
    logger.info("register Fn");
    Function txFunction = new MyTransactionFunction();
    FunctionService.registerFunction(txFunction);
  }

  private void runFunction(final Region pr, int cust, boolean isFirstFunc) {
    CustId custId = new CustId(cust);
    Customer newCus = new Customer("foo", "bar");
    ArrayList args = new ArrayList();
    Execution execution = FunctionService.onRegion(pr);
    Set filter = new HashSet();

    args.add(new Integer(UPDATE_NON_COLOCATION));
    logger.info("UPDATE_NON_COLOCATION");
    args.add(custId);
    args.add(newCus);
    filter.add(custId);
    try {
      execution.withFilter(filter).setArguments(args).execute(new MyTransactionFunction().getId())
          .getResult();
      assertTrue("Expected exception was not thrown", isFirstFunc);
    } catch (Exception exp) {
      if (!isFirstFunc) {
        if (exp instanceof TransactionException && exp.getMessage()
            .startsWith("Function execution is not colocated with transaction.")) {
        } else {
          logger.info("Expected to catch TransactionException but caught exception " + exp, exp);
          Assert.fail("Expected to catch TransactionException but caught exception ", exp);
        }
      } else {
        logger.info("Caught unexpected exception", exp);
        Assert.fail("Unexpected exception was thrown", exp);
      }
    }
  }

  private void runTXFunctions() {
    PartitionedRegion pr = (PartitionedRegion) basicGetCache()
        .getRegion(Region.SEPARATOR + CustomerPartitionedRegionName);
    CacheTransactionManager mgr = pr.getCache().getCacheTransactionManager();
    mgr.begin();
    runFunction(pr, 1, true);
    runFunction(pr, 2, false);
    mgr.commit();
  }

  /**
   * Test all the basic functionality of colocated transactions. This method invokes
   * {@link MyTransactionFunction} and tells it what to test, using different arguments.
   *
   * @param redundantBuckets redundant buckets for colocated PRs
   * @param populateData if false tests are carried out on empty colocated PRs
   */
  private void basicPRTXInFunction(int redundantBuckets, boolean populateData) {

    if (populateData) {
      createPopulateAndVerifyCoLocatedPRs(redundantBuckets);
    } else {
      createColocatedPRs(redundantBuckets);
    }

    dataStore1.invoke(() -> registerFunction());
    dataStore2.invoke(() -> registerFunction());
    dataStore3.invoke(() -> registerFunction());

    accessor.invoke(new SerializableCallable("run function") {
      @Override
      public Object call() {
        PartitionedRegion pr = (PartitionedRegion) basicGetCache()
            .getRegion(Region.SEPARATOR + CustomerPartitionedRegionName);
        PartitionedRegion orderpr = (PartitionedRegion) basicGetCache()
            .getRegion(Region.SEPARATOR + OrderPartitionedRegionName);
        CustId custId = new CustId(2);
        Customer newCus = new Customer("foo", "bar");
        Order order = new Order("fooOrder");
        OrderId orderId = new OrderId(22, custId);
        ArrayList args = new ArrayList();
        Function txFunction = new MyTransactionFunction();
        FunctionService.registerFunction(txFunction);
        Execution execution = FunctionService.onRegion(pr);
        Set filter = new HashSet();
        // test transaction non-coLocated operations
        filter.clear();
        args.clear();
        args.add(new Integer(VERIFY_NON_COLOCATION));
        LogWriterUtils.getLogWriter().info("VERIFY_NON_COLOCATION");
        args.add(custId);
        args.add(newCus);
        args.add(orderId);
        args.add(order);
        filter.add(custId);
        try {
          execution.withFilter(filter).setArguments(args).execute(txFunction.getId()).getResult();
          fail("Expected exception was not thrown");
        } catch (FunctionException fe) {
          LogWriterUtils.getLogWriter().info("Caught Expected exception");
          if (fe.getCause() instanceof TransactionDataNotColocatedException) {
          } else {
            throw new TestException(
                "Expected to catch FunctionException with cause TransactionDataNotColocatedException"
                    + " but cause is  " + fe.getCause(),
                fe.getCause());
          }
        }
        // verify that the transaction modifications are applied
        args.set(0, new Integer(VERIFY_TX));
        LogWriterUtils.getLogWriter().info("VERIFY_TX");
        orderpr.put(orderId, order);
        assertNotNull(orderpr.get(orderId));
        execution.withFilter(filter).setArguments(args).execute(txFunction.getId()).getResult();
        assertTrue("Unexpected customer value after commit", newCus.equals(pr.get(custId)));
        Order commitedOrder = (Order) orderpr.get(orderId);
        assertTrue(
            "Unexpected order value after commit. Expected:" + order + " Found:" + commitedOrder,
            order.equals(commitedOrder));
        // verify conflict detection
        args.set(0, new Integer(VERIFY_TXSTATE_CONFLICT));
        execution.withFilter(filter).setArguments(args).execute(txFunction.getId()).getResult();
        // verify that the transaction is rolled back
        args.set(0, new Integer(VERIFY_ROLLBACK));
        LogWriterUtils.getLogWriter().info("VERIFY_ROLLBACK");
        execution.withFilter(filter).setArguments(args).execute(txFunction.getId()).getResult();
        // verify destroy
        args.set(0, new Integer(VERIFY_DESTROY));
        LogWriterUtils.getLogWriter().info("VERIFY_DESTROY");
        execution.withFilter(filter).setArguments(args).execute(txFunction.getId()).getResult();
        // verify invalidate
        args.set(0, new Integer(VERIFY_INVALIDATE));
        LogWriterUtils.getLogWriter().info("VERIFY_INVALIDATE");
        execution.withFilter(filter).setArguments(args).execute(txFunction.getId()).getResult();
        return Boolean.TRUE;
      }
    });
  }

  private void createColocatedPRs(int redundantBuckets) {
    createCacheInAllVms();

    redundancy = new Integer(redundantBuckets);
    localMaxmemory = new Integer(50);
    totalNumBuckets = new Integer(11);

    // Create Customer PartitionedRegion in All VMs
    createPRWithCoLocation(CustomerPartitionedRegionName, null);

    // Create Order PartitionedRegion in All VMs
    createPRWithCoLocation(OrderPartitionedRegionName, CustomerPartitionedRegionName);

    // Create Shipment PartitionedRegion in All VMs
    createPRWithCoLocation(ShipmentPartitionedRegionName, OrderPartitionedRegionName);

    // Initial Validation for the number of data stores and number of profiles
    accessor.invoke(() -> PRColocationDUnitTest
        .validateBeforePutCustomerPartitionedRegion(CustomerPartitionedRegionName));
  }

  @Override
  protected void createCacheInAllVms() {
    dataStore1.invoke(runGetCache);
    dataStore2.invoke(runGetCache);
    dataStore3.invoke(runGetCache);
    accessor.invoke(runGetCache);
  }

  private SerializableCallable runGetCache = new SerializableCallable("runGetCache") {
    @Override
    public Object call() {
      getCache();
      return null;
    }
  };

  private void populateAndVerifyColocatedPRs(int redundantBuckets) {
    // Put the customer 1-10 in CustomerPartitionedRegion
    accessor.invoke(
        () -> PRColocationDUnitTest.putCustomerPartitionedRegion(CustomerPartitionedRegionName));

    // Put the order 1-10 for each Customer in OrderPartitionedRegion
    accessor
        .invoke(() -> PRColocationDUnitTest.putOrderPartitionedRegion(OrderPartitionedRegionName));

    // Put the shipment 1-10 for each order in ShipmentPartitionedRegion
    accessor.invoke(
        () -> PRColocationDUnitTest.putShipmentPartitionedRegion(ShipmentPartitionedRegionName));

    // for VM0 DataStore check the number of buckets created and the size of
    // bucket for all partitionedRegion
    Integer totalBucketsInDataStore1 = (Integer) dataStore1
        .invoke(() -> PRColocationDUnitTest.validateDataStore(CustomerPartitionedRegionName,
            OrderPartitionedRegionName, ShipmentPartitionedRegionName));

    // for VM1 DataStore check the number of buckets created and the size of
    // bucket for all partitionedRegion
    Integer totalBucketsInDataStore2 = (Integer) dataStore2
        .invoke(() -> PRColocationDUnitTest.validateDataStore(CustomerPartitionedRegionName,
            OrderPartitionedRegionName, ShipmentPartitionedRegionName));

    // for VM3 Datastore check the number of buckets created and the size of
    // bucket for all partitionedRegion
    Integer totalBucketsInDataStore3 = (Integer) dataStore3
        .invoke(() -> PRColocationDUnitTest.validateDataStore(CustomerPartitionedRegionName,
            OrderPartitionedRegionName, ShipmentPartitionedRegionName));

    // Check the total number of buckets created in all three Vms are equalto 30
    totalNumBucketsInTest = totalBucketsInDataStore1.intValue()
        + totalBucketsInDataStore2.intValue() + totalBucketsInDataStore3.intValue();
    assertEquals(30 + (redundantBuckets * 30), totalNumBucketsInTest);

    // This is the importatnt check. Checks that the colocated Customer,Order
    // and Shipment are in the same VM

    accessor.invoke(() -> PRColocationDUnitTest.validateAfterPutPartitionedRegion(
        CustomerPartitionedRegionName, OrderPartitionedRegionName, ShipmentPartitionedRegionName));

  }

  private void createPopulateAndVerifyCoLocatedPRs(int redundantBuckets) {

    createColocatedPRs(redundantBuckets);
    populateAndVerifyColocatedPRs(redundantBuckets);
  }

  private void createPRWithCoLocation(String prName, String coLocatedWith) {
    setAttributes(prName, coLocatedWith);
    createPartitionedRegion(attributeObjects);
  }

  private void setAttributes(String prName, String coLocatedWith) {
    this.regionName = prName;
    this.colocatedWith = coLocatedWith;
    this.isPartitionResolver = new Boolean(true);
    this.attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver, getEnableConcurrency()};
  }

  protected boolean getEnableConcurrency() {
    return false;
  }

  /**
   * This method executes a transaction with get on non colocated entries and expects the
   * transaction to fail with TransactionDataNotColocatedException.
   *
   * @param bucketRedundancy redundancy for the colocated PRs
   */
  private void basicPRTXWithNonColocatedGet(int bucketRedundancy) {
    setupColocatedRegions(bucketRedundancy);

    dataStore1.invoke(verifyNonColocated);
    dataStore2.invoke(verifyNonColocated);

    dataStore1.invoke(getTx);
  }

  @SuppressWarnings("unchecked")
  private void setupColocatedRegions(int bucketRedundancy) {
    dataStore1.invoke(runGetCache);
    dataStore2.invoke(runGetCache);
    redundancy = new Integer(bucketRedundancy);
    localMaxmemory = new Integer(50);
    totalNumBuckets = new Integer(2);

    createColocatedRegionsInTwoNodes();

    // Put the customer 1-2 in CustomerPartitionedRegion
    dataStore1.invoke(
        () -> PRColocationDUnitTest.putCustomerPartitionedRegion(CustomerPartitionedRegionName, 2));

    // Put the associated order in colocated OrderPartitionedRegion
    dataStore1.invoke(
        () -> PRColocationDUnitTest.putOrderPartitionedRegion(OrderPartitionedRegionName, 2));
  }

  private void createColocatedRegionsInTwoNodes() {
    setAttributes(CustomerPartitionedRegionName, null);
    createPRInTwoNodes();
    setAttributes(OrderPartitionedRegionName, CustomerPartitionedRegionName);
    createPRInTwoNodes();
  }


  @SuppressWarnings("serial")
  private SerializableRunnable verifyNonColocated = new SerializableRunnable("verifyNonColocated") {
    @Override
    public void run() throws PRLocallyDestroyedException, ForceReattemptException {
      containsKeyLocally();
    }
  };

  @SuppressWarnings("serial")
  private SerializableRunnable getTx = new SerializableRunnable("getTx") {
    @Override
    public void run() {
      performGetTx();
    }
  };


  @SuppressWarnings({"unchecked", "rawtypes"})
  private void containsKeyLocally() throws PRLocallyDestroyedException, ForceReattemptException {
    PartitionedRegion pr = (PartitionedRegion) basicGetCache()
        .getRegion(Region.SEPARATOR + CustomerPartitionedRegionName);

    CustId cust1 = new CustId(1);
    CustId cust2 = new CustId(2);
    int bucketId1 = pr.getKeyInfo(cust1).getBucketId();
    int bucketId2 = pr.getKeyInfo(cust2).getBucketId();

    List<Integer> localPrimaryBucketList = pr.getLocalPrimaryBucketsListTestOnly();
    Set localBucket1Keys;
    Set localBucket2Keys;
    assertTrue(localPrimaryBucketList.size() == 1);
    for (int bucketId : localPrimaryBucketList) {
      if (bucketId == bucketId1) {
        // primary bucket has cust1
        localBucket1Keys = pr.getDataStore().getKeysLocally(bucketId1, false);
        for (Object key : localBucket1Keys) {
          LogService.getLogger().info("local key set contains " + key);
        }
        assertTrue(localBucket1Keys.size() == 1);
      } else {
        localBucket2Keys = pr.getDataStore().getKeysLocally(bucketId2, false);
        for (Object key : localBucket2Keys) {
          LogService.getLogger().info("local key set contains " + key);
        }
        assertTrue(localBucket2Keys.size() == 1);
      }
    }
  }


  private void performGetTx() {
    PartitionedRegion pr = (PartitionedRegion) basicGetCache()
        .getRegion(Region.SEPARATOR + CustomerPartitionedRegionName);
    CacheTransactionManager mgr = pr.getCache().getCacheTransactionManager();
    CustId cust1 = new CustId(1);
    CustId cust2 = new CustId(2);
    boolean isCust1Local = isCust1Local(pr, cust1);

    // touch first get on remote node -- using TXStateStub
    Assertions.assertThatThrownBy(() -> getTx(!isCust1Local, mgr, pr, cust1, cust2))
        .isInstanceOf(TransactionDataNotColocatedException.class);

    // touch first get on local node-- using TXState
    Assertions.assertThatThrownBy(() -> getTx(isCust1Local, mgr, pr, cust1, cust2))
        .isInstanceOf(TransactionDataNotColocatedException.class);
  }

  @SuppressWarnings("unchecked")
  private boolean isCust1Local(PartitionedRegion pr, CustId cust1) {
    int bucketId1 = pr.getKeyInfo(cust1).getBucketId();
    List<Integer> localPrimaryBucketList = pr.getLocalPrimaryBucketsListTestOnly();
    assertTrue(localPrimaryBucketList.size() == 1);
    return (Integer) localPrimaryBucketList.get(0) == bucketId1;
  }

  private void getTx(boolean doCust1First, CacheTransactionManager mgr, PartitionedRegion pr,
      CustId cust1, CustId cust2) {
    CustId first = doCust1First ? cust1 : cust2;
    CustId second = !doCust1First ? cust1 : cust2;

    mgr.begin();
    boolean doRollback = true;
    try {
      pr.get(first);
      pr.get(second);
      doRollback = false;
    } finally {
      if (doRollback) {
        mgr.rollback();
      } else {
        mgr.commit();
      }
    }
  }

  /**
   * This method executes a transaction with operation on a key in a moved bucket, and expects
   * transaction to fail with TransactionDataRebalancedException.
   *
   * @param op which entry op to be executed
   * @param bucketRedundancy redundancy for the colocated PRs
   */
  @SuppressWarnings("unchecked")
  private void basicPRTXWithOpOnMovedBucket(Op op, int bucketRedundancy, DistributedMember dm1,
      DistributedMember dm2) {
    // First transaction.
    TransactionId txId = dataStore1.invoke(() -> beginTx(false));
    dataStore1.invoke(resumeTx(op, txId, dm1, dm2));

    // Second one. Will go through different path (using TXState or TXStateStub)
    txId = dataStore1.invoke(() -> beginTx(false));
    dataStore1.invoke(resumeTx(op, txId, dm1, dm2));
  }

  @SuppressWarnings({"rawtypes", "serial"})
  private SerializableCallable getDM() {
    return new SerializableCallable("getDM") {
      @Override
      public Object call() {
        return ((GemFireCacheImpl) basicGetCache()).getMyId();
      }
    };
  }

  private TransactionId beginTx(boolean doPut) {
    PartitionedRegion pr = (PartitionedRegion) basicGetCache()
        .getRegion(Region.SEPARATOR + CustomerPartitionedRegionName);
    CacheTransactionManager mgr = basicGetCache().getCacheTransactionManager();
    CustId cust1 = new CustId(1);
    mgr.begin();
    Object value = pr.get(cust1);
    assertNotNull(value);
    if (doPut) {
      pr.put(cust1, "bar");
    }
    return mgr.suspend();
  }

  @SuppressWarnings("serial")
  private SerializableRunnable resumeTx(Op op, TransactionId txId, DistributedMember dm1,
      DistributedMember dm2) {
    return new SerializableRunnable("resume tx") {
      @Override
      public void run() {
        PartitionedRegion pr = (PartitionedRegion) basicGetCache()
            .getRegion(Region.SEPARATOR + OrderPartitionedRegionName);
        CacheTransactionManager mgr = basicGetCache().getCacheTransactionManager();

        moveBucket(op, dm1, dm2);

        Assertions.assertThatThrownBy(() -> _resumeTx(op, txId, pr, mgr))
            .isInstanceOf(TransactionDataRebalancedException.class);
      }
    };
  }

  private void _resumeTx(Op op, TransactionId txId, PartitionedRegion pr,
      CacheTransactionManager mgr) {
    CustId cust1 = new CustId(1);
    OrderId order1 = new OrderId(11, cust1);
    OrderId neworder = new OrderId(111, cust1);
    mgr.resume(txId);
    try {
      switch (op) {
        case GET:
          pr.get(order1);
          break;
        case CONTAINSVALUEFORKEY:
          pr.containsValueForKey(order1);
          break;
        case CONTAINSKEY:
          pr.containsKey(order1);
          break;
        case CREATE:
          pr.create(neworder, new Order("test"));
          break;
        case PUT:
          pr.put(order1, new Order("test"));
          break;
        case INVALIDATE:
          pr.invalidate(order1);
          break;
        case DESTROY:
          pr.destroy(order1);
          break;
        case GETENTRY:
          pr.getEntry(order1);
          break;
        default:
          throw new AssertionError("Unknown operations " + op);
      }

    } finally {
      mgr.rollback();
    }
  }

  @SuppressWarnings("unchecked")
  private void moveBucket(Op op, DistributedMember dm1, DistributedMember dm2) {
    PartitionedRegion pr = (PartitionedRegion) basicGetCache()
        .getRegion(Region.SEPARATOR + CustomerPartitionedRegionName);
    CustId cust1 = new CustId(1);
    OrderId order1 = new OrderId(11, cust1);
    boolean isCust1Local = isCust1LocalSingleBucket(pr, cust1);
    DistributedMember source = isCust1Local ? dm1 : dm2;
    DistributedMember destination = isCust1Local ? dm2 : dm1;
    PartitionedRegion prOrder = (PartitionedRegion) basicGetCache()
        .getRegion(Region.SEPARATOR + OrderPartitionedRegionName);

    LogService.getLogger().info("source ={}, destination ={}", source, destination);

    switch (op) {
      case GET:
        moveBucketForGet(order1, isCust1Local, source, destination, prOrder);
        break;
      case CONTAINSVALUEFORKEY:
      case CONTAINSKEY:
      case CREATE:
      case PUT:
      case INVALIDATE:
      case DESTROY:
      case GETENTRY:
        PartitionRegionHelper.moveBucketByKey(prOrder, source, destination, order1);
        break;
      default:
        throw new AssertionError("Unknown operations " + op);
    }
  }

  @SuppressWarnings("unchecked")
  private void moveBucketForGet(OrderId order1, boolean isCust1Local, DistributedMember source,
      DistributedMember destination, PartitionedRegion prOrder) {
    if (isCust1Local && useBucketReadHook) {
      // Use TXState
      setBucketReadHook(order1, source, destination, prOrder);
    } else {
      // Use TXState and TXStateStub -- transaction data on remote node
      PartitionRegionHelper.moveBucketByKey(prOrder, source, destination, order1);
    }
  }

  private void setBucketReadHook(OrderId order1, DistributedMember source,
      DistributedMember destination, PartitionedRegion prOrder) {
    prOrder.getDataStore().setBucketReadHook(new Runnable() {
      @Override
      @SuppressWarnings("unchecked")
      public void run() {
        LogService.getLogger().info("In bucketReadHook");
        PartitionRegionHelper.moveBucketByKey(prOrder, source, destination, order1);
      }
    });
  }

  private void createPRInTwoNodes() {
    dataStore1.invoke(PRColocationDUnitTest.class, "createPR", this.attributeObjects);
    dataStore2.invoke(PRColocationDUnitTest.class, "createPR", this.attributeObjects);
  }

  @SuppressWarnings("unchecked")
  private boolean isCust1LocalSingleBucket(PartitionedRegion pr, CustId cust1) {
    List<Integer> localPrimaryBucketList = pr.getLocalPrimaryBucketsListTestOnly();
    return (Integer) localPrimaryBucketList.size() == 1;
  }

  @SuppressWarnings("unchecked")
  private void setupMoveBucket(int bucketRedundancy) {
    dataStore1.invoke(runGetCache);
    dataStore2.invoke(runGetCache);
    redundancy = new Integer(bucketRedundancy);
    localMaxmemory = new Integer(50);
    totalNumBuckets = new Integer(1);

    createColocatedRegionsInTwoNodes();

    // Put the customer 1 in CustomerPartitionedRegion
    dataStore1.invoke(
        () -> PRColocationDUnitTest.putCustomerPartitionedRegion(CustomerPartitionedRegionName, 1));

    // Put the associated order in colocated OrderPartitionedRegion
    dataStore1.invoke(
        () -> PRColocationDUnitTest.putOrderPartitionedRegion(OrderPartitionedRegionName, 1));
  }

  @Test
  public void testTxWithNonColocatedGet() {
    basicPRTXWithNonColocatedGet(0);
  }

  /**
   * This method executes a transaction with entry op on non colocated entries and expects the
   * transaction to fail with TransactionDataNotColocatedException.
   *
   * @param bucketRedundancy redundancy for the colocated PRs
   */
  private void basicPRTXWithNonColocatedOp(Op op, int bucketRedundancy) {
    dataStore1.invoke(verifyNonColocated);
    dataStore2.invoke(verifyNonColocated);

    // TXState and TXStateStub will be exectued on different nodes.
    dataStore1.invoke(doOp(op));
    dataStore2.invoke(doOp(op));
  }

  @SuppressWarnings("serial")
  private SerializableRunnable doOp(Op op) {
    return new SerializableRunnable("doOp") {
      @Override
      public void run() {
        basicDoOp(op);
      }
    };
  }

  private void basicDoOp(Op op) {
    CacheTransactionManager mgr = basicGetCache().getCacheTransactionManager();

    Assertions.assertThatThrownBy(() -> doOp(op, mgr))
        .isInstanceOf(TransactionDataNotColocatedException.class);
  }

  private void doOp(Op op, CacheTransactionManager mgr) {
    PartitionedRegion cust = (PartitionedRegion) basicGetCache()
        .getRegion(Region.SEPARATOR + CustomerPartitionedRegionName);
    PartitionedRegion order = (PartitionedRegion) basicGetCache()
        .getRegion(Region.SEPARATOR + OrderPartitionedRegionName);

    CustId cust1 = new CustId(1);
    CustId cust2 = new CustId(2);
    OrderId order2 = new OrderId(21, cust2);
    OrderId neworder2 = new OrderId(221, cust2);

    mgr.begin();
    cust.get(cust1); // touch 1 bucket

    try {
      switch (op) {
        case GET:
          order.get(order2);
          break;
        case CONTAINSVALUEFORKEY:
          order.containsValueForKey(order2);
          break;
        case CONTAINSKEY:
          order.containsKey(order2);
          break;
        case CREATE:
          order.create(neworder2, new Order("test"));
          break;
        case PUT:
          order.put(order2, new Order("test"));
          break;
        case INVALIDATE:
          order.invalidate(order2);
          break;
        case DESTROY:
          order.destroy(order2);
          break;
        case GETENTRY:
          order.getEntry(order2);
          break;
        default:
          throw new AssertionError("Unknown operations " + op);
      }
    } finally {
      mgr.rollback();
    }
  }

  @Test
  public void testTxWithNonColocatedOps() {
    setupColocatedRegions(0);

    for (Op op : Op.values()) {
      LogService.getLogger().info("Testing Operation: " + op.name());
      basicPRTXWithNonColocatedOp(op, 0);
    }
  }

  private boolean useBucketReadHook;

  @SuppressWarnings("unchecked")
  @Test
  public void testTxWithOpsOnMovedBucket() {
    setupMoveBucket(0);

    DistributedMember dm1 = (DistributedMember) dataStore1.invoke(getDM());
    DistributedMember dm2 = (DistributedMember) dataStore2.invoke(getDM());

    for (Op op : Op.values()) {
      LogService.getLogger().info("Testing Operation: " + op.name());
      basicPRTXWithOpOnMovedBucket(op, 0, dm1, dm2);
    }
  }

  @SuppressWarnings("unchecked")
  @Test
  public void testTxWithGetOnMovedBucketUsingBucketReadHook() {
    setupMoveBucket(0);

    DistributedMember dm1 = (DistributedMember) dataStore1.invoke(getDM());
    DistributedMember dm2 = (DistributedMember) dataStore2.invoke(getDM());

    Op op = Op.GET;
    useBucketReadHook = true;
    basicPRTXWithOpOnMovedBucket(op, 0, dm1, dm2);
    useBucketReadHook = false;
  }

  @Test
  public void testPRTXInCacheListenerRedundancy0() {
    basicPRTXInCacheListener(0);
  }

  @Test
  public void testPRTXInCacheListenerRedundancy1() {
    basicPRTXInCacheListener(1);
  }

  @Test
  public void testPRTXInCacheListenerRedundancy2() {
    basicPRTXInCacheListener(2);
  }

  /**
   * This method executes a transaction inside a cache listener
   *
   * @param bucketRedundancy redundancy for the colocated PRs
   */
  private void basicPRTXInCacheListener(int bucketRedundancy) {
    createCacheInAllVms();
    redundancy = new Integer(bucketRedundancy);
    localMaxmemory = new Integer(50);
    totalNumBuckets = new Integer(11);
    // Create Customer PartitionedRegion in All VMs
    createPRWithCoLocation(CustomerPartitionedRegionName, null);
    createPRWithCoLocation(OrderPartitionedRegionName, CustomerPartitionedRegionName);
    // register Cache Listeners
    SerializableCallable registerListeners = new SerializableCallable() {
      @Override
      public Object call() {
        Region custRegion =
            basicGetCache().getRegion(Region.SEPARATOR + CustomerPartitionedRegionName);
        custRegion.getAttributesMutator().addCacheListener(new TransactionListener());
        return null;
      }
    };

    dataStore1.invoke(registerListeners);
    dataStore2.invoke(registerListeners);
    dataStore3.invoke(registerListeners);

    // Put the customer 1-10 in CustomerPartitionedRegion
    accessor.invoke(
        () -> PRColocationDUnitTest.putCustomerPartitionedRegion(CustomerPartitionedRegionName));

    dataStore1.invoke(() -> PRTransactionDUnitTest.validatePRTXInCacheListener());
    dataStore2.invoke(() -> PRTransactionDUnitTest.validatePRTXInCacheListener());
    dataStore3.invoke(() -> PRTransactionDUnitTest.validatePRTXInCacheListener());

  }

  /**
   * verify that 10 orders are created for each customer
   *
   */
  private static void validatePRTXInCacheListener() {
    PartitionedRegion customerPartitionedregion = null;
    PartitionedRegion orderPartitionedregion = null;
    try {
      customerPartitionedregion = (PartitionedRegion) basicGetCache()
          .getRegion(Region.SEPARATOR + CustomerPartitionedRegionName);
      orderPartitionedregion = (PartitionedRegion) basicGetCache()
          .getRegion(Region.SEPARATOR + OrderPartitionedRegionName);
    } catch (Exception e) {
      Assert.fail("validateAfterPutPartitionedRegion : failed while getting the region", e);
    }
    assertNotNull(customerPartitionedregion);
    assertNotNull(orderPartitionedregion);

    customerPartitionedregion.getDataStore().dumpEntries(false);
    orderPartitionedregion.getDataStore().dumpEntries(false);
    Iterator custIterator = customerPartitionedregion.getDataStore().getEntries().iterator();
    LogWriterUtils.getLogWriter()
        .info("Found " + customerPartitionedregion.getDataStore().getEntries().size()
            + " Customer entries in the partition");
    Region.Entry custEntry = null;
    while (custIterator.hasNext()) {
      custEntry = (Entry) custIterator.next();
      CustId custid = (CustId) custEntry.getKey();
      Customer cust = (Customer) custEntry.getValue();
      Iterator orderIterator = orderPartitionedregion.getDataStore().getEntries().iterator();
      LogWriterUtils.getLogWriter()
          .info("Found " + orderPartitionedregion.getDataStore().getEntries().size()
              + " Order entries in the partition");
      int orderPerCustomer = 0;
      Region.Entry orderEntry = null;
      while (orderIterator.hasNext()) {
        orderEntry = (Entry) orderIterator.next();
        OrderId orderId = (OrderId) orderEntry.getKey();
        Order order = (Order) orderEntry.getValue();
        if (custid.equals(orderId.getCustId())) {
          orderPerCustomer++;
        }
      }
      assertEquals(10, orderPerCustomer);
    }
  }

  @Ignore("BUG46661")
  @Test
  public void testCacheListenerCallbacks() {
    createPopulateAndVerifyCoLocatedPRs(1);

    SerializableCallable registerListeners = new SerializableCallable() {
      @Override
      public Object call() {
        Region custRegion =
            basicGetCache().getRegion(Region.SEPARATOR + CustomerPartitionedRegionName);
        custRegion.getAttributesMutator().addCacheListener(new TransactionListener2());
        return null;
      }
    };

    accessor.invoke(registerListeners);
    dataStore1.invoke(registerListeners);
    dataStore2.invoke(registerListeners);
    dataStore3.invoke(registerListeners);

    accessor.invoke(new SerializableCallable("run function") {
      @Override
      public Object call() {
        PartitionedRegion pr = (PartitionedRegion) basicGetCache()
            .getRegion(Region.SEPARATOR + CustomerPartitionedRegionName);
        PartitionedRegion orderpr = (PartitionedRegion) basicGetCache()
            .getRegion(Region.SEPARATOR + OrderPartitionedRegionName);
        CustId custId = new CustId(2);
        Customer newCus = new Customer("foo", "bar");
        Order order = new Order("fooOrder");
        OrderId orderId = new OrderId(22, custId);
        ArrayList args = new ArrayList();
        Function txFunction = new MyTransactionFunction();
        FunctionService.registerFunction(txFunction);
        Execution execution = FunctionService.onRegion(pr);
        Set filter = new HashSet();
        boolean caughtException = false;
        // test transaction non-coLocated operations
        filter.clear();
        args.clear();
        args.add(new Integer(VERIFY_LISTENER_CALLBACK));
        LogWriterUtils.getLogWriter().info("VERIFY_LISTENER_CALLBACK");
        args.add(custId);
        args.add(newCus);
        args.add(orderId);
        args.add(order);
        filter.add(custId);
        caughtException = false;
        execution.withFilter(filter).setArguments(args).execute(txFunction.getId()).getResult();
        return null;
      }
    });

  }

  @Test
  public void testRepeatableRead() {
    createColocatedPRs(1);

    dataStore1.invoke(() -> registerFunction());
    dataStore2.invoke(() -> registerFunction());
    dataStore3.invoke(() -> registerFunction());

    accessor.invoke(new SerializableCallable("run function") {
      @Override
      public Object call() {
        PartitionedRegion pr = (PartitionedRegion) basicGetCache()
            .getRegion(Region.SEPARATOR + CustomerPartitionedRegionName);
        PartitionedRegion orderpr = (PartitionedRegion) basicGetCache()
            .getRegion(Region.SEPARATOR + OrderPartitionedRegionName);
        CustId custId = new CustId(2);
        Customer newCus = new Customer("foo", "bar");
        Order order = new Order("fooOrder");
        OrderId orderId = new OrderId(22, custId);
        ArrayList args = new ArrayList();
        Function txFunction = new MyTransactionFunction();
        FunctionService.registerFunction(txFunction);
        Execution execution = FunctionService.onRegion(pr);
        Set filter = new HashSet();
        filter.clear();
        args.clear();
        args.add(new Integer(VERIFY_REP_READ));
        LogWriterUtils.getLogWriter().info("VERIFY_REP_READ");
        args.add(custId);
        args.add(newCus);
        args.add(orderId);
        args.add(order);
        filter.add(custId);
        execution.withFilter(filter).setArguments(args).execute(txFunction.getId()).getResult();

        return null;
      }
    });

  }

  @Test
  public void testPRTXPerformance() {
    PRColocationDUnitTestHelper.defaultStringSize = 1024;

    createPopulateAndVerifyCoLocatedPRs(1);

    // register functions
    SerializableCallable registerPerfFunctions = new SerializableCallable("register Fn") {
      @Override
      public Object call() {
        Function perfFunction = new PerfFunction();
        FunctionService.registerFunction(perfFunction);
        Function perfTxFunction = new PerfTxFunction();
        FunctionService.registerFunction(perfTxFunction);
        return Boolean.TRUE;
      }
    };

    dataStore1.invoke(registerPerfFunctions);
    dataStore2.invoke(registerPerfFunctions);
    dataStore3.invoke(registerPerfFunctions);
    accessor.invoke(registerPerfFunctions);

    SerializableCallable runPerfFunction = new SerializableCallable("runPerfFunction") {
      @Override
      public Object call() {
        long perfTime = 0;
        Region customerPR = basicGetCache().getRegion(CustomerPartitionedRegionName);
        Execution execution = FunctionService.onRegion(customerPR);
        // for each customer, update order and shipment
        for (int iterations = 1; iterations <= totalIterations; iterations++) {
          LogWriterUtils.getLogWriter().info("running perfFunction");
          long startTime = 0;
          ArrayList args = new ArrayList();
          CustId custId = new CustId(iterations % 10);
          for (int i = 1; i <= perfOrderShipmentPairs; i++) {
            OrderId orderId = new OrderId(custId.getCustId().intValue() * 10 + i, custId);
            Order order = new Order("NewOrder" + i + iterations);
            ShipmentId shipmentId =
                new ShipmentId(orderId.getOrderId().intValue() * 10 + i, orderId);
            Shipment shipment = new Shipment("newShipment" + i + iterations);
            args.add(orderId);
            args.add(order);
            args.add(shipmentId);
            args.add(shipment);
          }
          Set filter = new HashSet();
          filter.add(custId);
          if (iterations > warmupIterations) {
            startTime = NanoTimer.getTime();
          }
          execution.withFilter(filter).setArguments(args).execute("perfFunction").getResult();
          if (startTime > 0) {
            perfTime += NanoTimer.getTime() - startTime;
          }
        }
        return new Long(perfTime);
      }
    };

    Long perfTime = (Long) accessor.invoke(runPerfFunction);

    SerializableCallable runPerfTxFunction = new SerializableCallable("runPerfTxFunction") {
      @Override
      public Object call() {
        long perfTime = 0;
        Region customerPR = basicGetCache().getRegion(CustomerPartitionedRegionName);
        Execution execution = FunctionService.onRegion(customerPR);
        // for each customer, update order and shipment
        for (int iterations = 1; iterations <= totalIterations; iterations++) {
          LogWriterUtils.getLogWriter().info("Running perfFunction");
          long startTime = 0;
          ArrayList args = new ArrayList();
          CustId custId = new CustId(iterations % 10);
          for (int i = 1; i <= perfOrderShipmentPairs; i++) {
            OrderId orderId = new OrderId(custId.getCustId().intValue() * 10 + i, custId);
            Order order = new Order("NewOrder" + i + iterations);
            ShipmentId shipmentId =
                new ShipmentId(orderId.getOrderId().intValue() * 10 + i, orderId);
            Shipment shipment = new Shipment("newShipment" + i + iterations);
            args.add(orderId);
            args.add(order);
            args.add(shipmentId);
            args.add(shipment);
          }
          Set filter = new HashSet();
          filter.add(custId);
          if (iterations > warmupIterations) {
            startTime = NanoTimer.getTime();
          }
          execution.withFilter(filter).setArguments(args).execute("perfTxFunction").getResult();
          if (startTime > 0) {
            perfTime += NanoTimer.getTime() - startTime;
          }
        }
        return new Long(perfTime);
      }
    };

    Long perfTxTime = (Long) accessor.invoke(runPerfTxFunction);

    double diff = (perfTime.longValue() - perfTxTime.longValue()) * 1.0;
    double percentDiff = (diff / perfTime.longValue()) * 100;

    LogWriterUtils.getLogWriter()
        .info((totalIterations - warmupIterations) + " iterations of function took:"
            + +perfTime.longValue() + " Nanos, and transaction function took:"
            + perfTxTime.longValue() + " Nanos, difference :" + diff + " percentDifference:"
            + percentDiff);

  }

  @Test
  public void testCommitToFailAfterPrimaryBucketMoved() {
    basicPRTXCommitToFailAfterPrimaryBucketMoved(1);
  }

  /**
   * Test commit fail after the transactional node no longer hosts the primary bucket of the
   * operations executed in the transaction.
   *
   * @param redundantBuckets redundant buckets for colocated PRs
   */
  private void basicPRTXCommitToFailAfterPrimaryBucketMoved(int redundantBuckets) {
    setupColocatedRegions(redundantBuckets);

    InternalDistributedMember dm1 = (InternalDistributedMember) dataStore1.invoke(getDM());
    InternalDistributedMember dm2 = (InternalDistributedMember) dataStore2.invoke(getDM());

    TransactionId txId = dataStore1.invoke(() -> beginTx(true));
    dataStore1.invoke(() -> movePrimaryBucket(dm1, dm2));
    dataStore1.invoke(() -> resumeTxAfterPrimaryMoved(txId));

  }

  private void movePrimaryBucket(InternalDistributedMember dm1, InternalDistributedMember dm2) {
    PartitionedRegion pr = (PartitionedRegion) basicGetCache()
        .getRegion(Region.SEPARATOR + CustomerPartitionedRegionName);
    CustId cust1 = new CustId(1);
    int bucketId = pr.getKeyInfo(cust1).getBucketId();
    boolean isCust1LocalPrimary = pr.getBucketRegion(cust1).getBucketAdvisor().isPrimary();
    InternalDistributedMember destination = isCust1LocalPrimary ? dm2 : dm1;
    InternalDistributedMember source = isCust1LocalPrimary ? dm1 : dm2;

    ExplicitMoveDirector director = new ExplicitMoveDirector(cust1, bucketId, source, destination,
        pr.getCache().getDistributedSystem());
    PartitionedRegionRebalanceOp rebalanceOp =
        new PartitionedRegionRebalanceOp(pr, false, director, true, true);
    BucketOperatorImpl operator = new BucketOperatorImpl(rebalanceOp);
    boolean moved = operator.movePrimary(source, destination, bucketId);
    if (!moved) {
      fail("Not able to move primary bucket by invoking BucketOperatorImpl.movePrimary");
    }
  }

  private void resumeTxAfterPrimaryMoved(TransactionId txId) {
    PartitionedRegion pr = (PartitionedRegion) basicGetCache()
        .getRegion(Region.SEPARATOR + CustomerPartitionedRegionName);
    CacheTransactionManager mgr = basicGetCache().getCacheTransactionManager();

    mgr.resume(txId);

    Throwable thrown = catchThrowable(() -> mgr.commit());
    assertTrue(thrown instanceof TransactionDataRebalancedException);
  }

  // Don't want to run the test twice
  @Override
  @Test
  public void testColocatedPartitionedRegion() {}

  @Override
  @Test
  public void testColocationPartitionedRegion() {}

  @Override
  @Test
  public void testColocationPartitionedRegionWithRedundancy() {}

  @Override
  @Test
  public void testPartitionResolverPartitionedRegion() {}

  @Override
  @Test
  public void testColocationPartitionedRegionWithNullColocationSpecifiedOnOneNode() {}

  @Override
  @Test
  public void testColocatedPRRedundancyRecovery() {}

  @Override
  @Test
  public void testColocatedPRWithAccessorOnDifferentNode1() {}

  @Override
  @Test
  public void testColocatedPRWithAccessorOnDifferentNode2() {}

  @Override
  @Test
  public void testColocatedPRWithDestroy() {}

  @Override
  @Test
  public void testColocatedPRWithLocalDestroy() {}

  @Override
  @Test
  public void testColocatedPRWithPROnDifferentNode1() {}

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

  private SerializableCallable verifyNoTxState = new SerializableCallable() {
    @Override
    public Object call() {
      TXManagerImpl mgr = getGemfireCache().getTxManager();
      assertEquals(0, mgr.hostedTransactionsInProgressForTest());
      return null;
    }
  };

  enum Op {
    GET, CONTAINSVALUEFORKEY, CONTAINSKEY, CREATE, PUT, INVALIDATE, DESTROY, GETENTRY;
  }

  static class TransactionListener extends CacheListenerAdapter {

    @Override
    public void afterCreate(EntryEvent event) {
      // for each customer, put 10 orders in a tx.
      Region custPR = event.getRegion();
      Region orderPR = custPR.getCache().getRegion(Region.SEPARATOR + OrderPartitionedRegionName);
      CacheTransactionManager mgr = custPR.getCache().getCacheTransactionManager();
      mgr.begin();
      CustId custId = (CustId) event.getKey();
      for (int j = 1; j <= 10; j++) {
        int oid = (custId.getCustId().intValue() * 10) + j;
        OrderId orderId = new OrderId(oid, custId);
        Order order = new Order("OREDR" + oid);
        try {
          assertNotNull(orderPR);
          orderPR.put(orderId, order);
        } catch (Exception e) {
          // mgr.rollback();
          Assert.fail(" failed while doing put operation in CacheListener ", e);
        }
      }
      mgr.commit();
      LogWriterUtils.getLogWriter().info("COMMIT completed");
    }
  }

  static class TransactionListener2 extends CacheListenerAdapter {

    private int numberOfPutCallbacks;
    private int numberOfDestroyCallbacks;
    private int numberOfInvalidateCallbacks;

    @Override
    public void afterCreate(EntryEvent event) {
      synchronized (this) {
        numberOfPutCallbacks++;
      }
    }

    @Override
    public void afterUpdate(EntryEvent event) {
      afterCreate(event);
    }

    @Override
    public void afterDestroy(EntryEvent event) {
      synchronized (this) {
        numberOfDestroyCallbacks++;
      }
    }

    @Override
    public void afterInvalidate(EntryEvent event) {
      synchronized (this) {
        numberOfInvalidateCallbacks++;
      }
    }

    synchronized int getNumberOfPutCallbacks() {
      return numberOfPutCallbacks;
    }

    synchronized int getNumberOfDestroyCallbacks() {
      return numberOfDestroyCallbacks;
    }

    synchronized int getNumberOfInvalidateCallbacks() {
      return numberOfInvalidateCallbacks;
    }
  }
}