/*
 * 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.apache.geode.cache.Region.SEPARATOR;
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.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
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 org.junit.Test;

import org.apache.geode.DataSerializable;
import org.apache.geode.DataSerializer;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.EntryOperation;
import org.apache.geode.cache.PartitionAttributes;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.PartitionResolver;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.execute.Execution;
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.execute.ResultCollector;
import org.apache.geode.cache30.CacheSerializableRunnable;
import org.apache.geode.distributed.ConfigurationProperties;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.ColocationHelper;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.control.InternalResourceManager;
import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceObserverAdapter;
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.test.dunit.Assert;
import org.apache.geode.test.dunit.AsyncInvocation;
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.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.Wait;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;

/**
 * This is the test for the custom and colocated partitioning of PartitionedRegion
 *
 */
@SuppressWarnings("synthetic-access")

public class PRColocationDUnitTest extends JUnit4CacheTestCase {

  VM accessor = null;

  VM dataStore1 = null;

  VM dataStore2 = null;

  VM dataStore3 = null;

  protected static Cache cache = null;

  protected static int totalNumBucketsInTest = 0;

  static final String CustomerPartitionedRegionName = "CustomerPartitionedRegion";

  static final String OrderPartitionedRegionName = "OrderPartitionedRegion";

  static final String ShipmentPartitionedRegionName = "ShipmentPartitionedRegion";

  String regionName = null;

  Integer redundancy = null;

  Integer localMaxmemory = null;

  Integer totalNumBuckets = null;

  String colocatedWith = null;

  Boolean isPartitionResolver = null;

  Object[] attributeObjects = null;

  public PRColocationDUnitTest() {
    super();
  }

  @Override
  public final void postSetUp() throws Exception {
    disconnectAllFromDS(); // isolate this test from others to avoid periodic CacheExistsExceptions
    Host host = Host.getHost(0);
    dataStore1 = host.getVM(0);
    dataStore2 = host.getVM(1);
    dataStore3 = host.getVM(2);
    accessor = host.getVM(3);
  }

  /*
   * Test for bug 41820
   */
  @Test
  public void testDestroyColocatedPartitionedRegion() throws Throwable {
    createCacheInAllVms();
    redundancy = 0;
    localMaxmemory = 50;
    totalNumBuckets = 11;

    regionName = "A";
    colocatedWith = null;
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "B";
    colocatedWith = SEPARATOR + "A";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "C";
    colocatedWith = SEPARATOR + "A";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    attributeObjects = new Object[] {SEPARATOR + "A"};
    dataStore1.invoke(PRColocationDUnitTest.class, "destroyPR", attributeObjects);
  }

  /*
   * Test for checking the colocation of the regions which forms the tree
   */
  @Test
  public void testColocatedPartitionedRegion() throws Throwable {
    createCacheInAllVms();
    redundancy = 0;
    localMaxmemory = 50;
    totalNumBuckets = 11;

    regionName = "A";
    colocatedWith = null;
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "B";
    colocatedWith = SEPARATOR + "A";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "C";
    colocatedWith = SEPARATOR + "A";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "D";
    colocatedWith = SEPARATOR + "B";
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "E";
    colocatedWith = SEPARATOR + "B";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "F";
    colocatedWith = SEPARATOR + "B";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "G";
    colocatedWith = SEPARATOR + "C";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "H";
    colocatedWith = SEPARATOR + "C";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "I";
    colocatedWith = SEPARATOR + "C";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "J";
    colocatedWith = SEPARATOR + "D";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "K";
    colocatedWith = SEPARATOR + "D";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "L";
    colocatedWith = SEPARATOR + "E";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "M";
    colocatedWith = SEPARATOR + "F";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "N";
    colocatedWith = SEPARATOR + "G";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "O";
    colocatedWith = SEPARATOR + "I";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    accessor.invoke(() -> PRColocationDUnitTest.validateColocatedRegions("A"));

    accessor.invoke(() -> PRColocationDUnitTest.validateColocatedRegions("D"));

    accessor.invoke(() -> PRColocationDUnitTest.validateColocatedRegions("H"));

    accessor.invoke(() -> PRColocationDUnitTest.validateColocatedRegions("B"));

    accessor.invoke(() -> PRColocationDUnitTest.validateColocatedRegions("K"));
  }

  /*
   * Test for checking the colocation of the regions which forms the tree
   */
  @Test
  public void testColocatedPartitionedRegion_NoFullPath() throws Throwable {
    createCacheInAllVms();
    redundancy = 0;
    localMaxmemory = 50;
    totalNumBuckets = 11;

    regionName = "A";
    colocatedWith = null;
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "B";
    colocatedWith = "A";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "C";
    colocatedWith = "A";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "D";
    colocatedWith = "B";
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "E";
    colocatedWith = "B";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "F";
    colocatedWith = "B";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "G";
    colocatedWith = "C";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "H";
    colocatedWith = "C";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "I";
    colocatedWith = "C";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "J";
    colocatedWith = "D";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "K";
    colocatedWith = "D";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "L";
    colocatedWith = "E";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "M";
    colocatedWith = "F";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "N";
    colocatedWith = "G";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    regionName = "O";
    colocatedWith = "I";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createPartitionedRegion(attributeObjects);

    accessor.invoke(() -> PRColocationDUnitTest.validateColocatedRegions("A"));

    accessor.invoke(() -> PRColocationDUnitTest.validateColocatedRegions("D"));

    accessor.invoke(() -> PRColocationDUnitTest.validateColocatedRegions("H"));

    accessor.invoke(() -> PRColocationDUnitTest.validateColocatedRegions("B"));

    accessor.invoke(() -> PRColocationDUnitTest.validateColocatedRegions("K"));
  }

  @Test
  public void testColocatedSubPartitionedRegion() throws Throwable {
    createCacheInAllVms();
    redundancy = 1;
    localMaxmemory = 50;
    totalNumBuckets = 11;

    regionName = "A";
    colocatedWith = null;
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "B";
    colocatedWith = SEPARATOR + "rootA" + SEPARATOR + "A";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "C";
    colocatedWith = SEPARATOR + "rootA" + SEPARATOR + "A";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "D";
    colocatedWith = SEPARATOR + "rootB" + SEPARATOR + "B";
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "E";
    colocatedWith = SEPARATOR + "rootB" + SEPARATOR + "B";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "F";
    colocatedWith = SEPARATOR + "rootB" + SEPARATOR + "B";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "G";
    colocatedWith = SEPARATOR + "rootC" + SEPARATOR + "C";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "H";
    colocatedWith = SEPARATOR + "rootC" + SEPARATOR + "C";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "I";
    colocatedWith = SEPARATOR + "rootC" + SEPARATOR + "C";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "J";
    colocatedWith = SEPARATOR + "rootD" + SEPARATOR + "D";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "K";
    colocatedWith = SEPARATOR + "rootD" + SEPARATOR + "D";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "L";
    colocatedWith = SEPARATOR + "rootE" + SEPARATOR + "E";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "M";
    colocatedWith = SEPARATOR + "rootF" + SEPARATOR + "F";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "N";
    colocatedWith = SEPARATOR + "rootG" + SEPARATOR + "G";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "O";
    colocatedWith = SEPARATOR + "rootI" + SEPARATOR + "I";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    accessor
        .invoke(() -> PRColocationDUnitTest.validateColocatedRegions("rootA" + SEPARATOR + "A"));

    accessor
        .invoke(() -> PRColocationDUnitTest.validateColocatedRegions("rootD" + SEPARATOR + "D"));

    accessor
        .invoke(() -> PRColocationDUnitTest.validateColocatedRegions("rootH" + SEPARATOR + "H"));

    accessor
        .invoke(() -> PRColocationDUnitTest.validateColocatedRegions("rootB" + SEPARATOR + "B"));

    accessor
        .invoke(() -> PRColocationDUnitTest.validateColocatedRegions("rootK" + SEPARATOR + "K"));
  }

  @Test
  public void testColocatedSubPartitionedRegion_NoFullPath() throws Throwable {
    createCacheInAllVms();
    redundancy = 1;
    localMaxmemory = 50;
    totalNumBuckets = 11;

    regionName = "A";
    colocatedWith = null;
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "B";
    colocatedWith = "rootA" + SEPARATOR + "A";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "C";
    colocatedWith = "rootA" + SEPARATOR + "A";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "D";
    colocatedWith = "rootB" + SEPARATOR + "B";
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "E";
    colocatedWith = "rootB" + SEPARATOR + "B";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "F";
    colocatedWith = "rootB" + SEPARATOR + "B";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "G";
    colocatedWith = "rootC" + SEPARATOR + "C";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "H";
    colocatedWith = "rootC" + SEPARATOR + "C";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "I";
    colocatedWith = "rootC" + SEPARATOR + "C";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "J";
    colocatedWith = "rootD" + SEPARATOR + "D";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "K";
    colocatedWith = "rootD" + SEPARATOR + "D";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "L";
    colocatedWith = "rootE" + SEPARATOR + "E";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "M";
    colocatedWith = "rootF" + SEPARATOR + "F";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "N";
    colocatedWith = "rootG" + SEPARATOR + "G";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    regionName = "O";
    colocatedWith = "rootI" + SEPARATOR + "I";
    isPartitionResolver = Boolean.TRUE;
    attributeObjects = new Object[] {regionName, redundancy, localMaxmemory, totalNumBuckets,
        colocatedWith, isPartitionResolver};
    createSubPartitionedRegion(attributeObjects);

    accessor
        .invoke(() -> PRColocationDUnitTest.validateColocatedRegions("rootA" + SEPARATOR + "A"));

    accessor
        .invoke(() -> PRColocationDUnitTest.validateColocatedRegions("rootD" + SEPARATOR + "D"));

    accessor
        .invoke(() -> PRColocationDUnitTest.validateColocatedRegions("rootH" + SEPARATOR + "H"));

    accessor
        .invoke(() -> PRColocationDUnitTest.validateColocatedRegions("rootB" + SEPARATOR + "B"));

    accessor
        .invoke(() -> PRColocationDUnitTest.validateColocatedRegions("rootK" + SEPARATOR + "K"));
  }

  @Test
  public void testColocatedPRWithAccessorOnDifferentNode1() throws Throwable {

    createCacheInAllVms();

    dataStore1.invoke(new CacheSerializableRunnable("testColocatedPRwithAccessorOnDifferentNode") {
      @Override
      public void run2() {
        String partitionedRegionName = CustomerPartitionedRegionName;
        colocatedWith = null;
        isPartitionResolver = Boolean.FALSE;
        redundancy = 0;
        localMaxmemory = 50;
        totalNumBuckets = 11;

        PartitionAttributesFactory paf = new PartitionAttributesFactory();
        paf.setRedundantCopies(redundancy).setLocalMaxMemory(localMaxmemory)
            .setTotalNumBuckets(totalNumBuckets).setColocatedWith(colocatedWith);
        if (isPartitionResolver) {
          paf.setPartitionResolver(new CustomerIDPartitionResolver("CustomerIDPartitionResolver"));
        }
        PartitionAttributes prAttr = paf.create();
        AttributesFactory attr = new AttributesFactory();
        attr.setPartitionAttributes(prAttr);
        assertNotNull(basicGetCache());
        Region pr = basicGetCache().createRegion(partitionedRegionName, attr.create());
        assertNotNull(pr);
        LogWriterUtils.getLogWriter().info("Partitioned Region " + partitionedRegionName
            + " created Successfully :" + pr);
      }
    });

    // add expected exception string
    final IgnoredException ex = IgnoredException.addIgnoredException(
        "Colocated regions should have accessors at the same node", dataStore1);
    dataStore1
        .invoke(new CacheSerializableRunnable("Colocated PR with Accessor on different nodes") {
          @Override
          public void run2() {
            regionName = OrderPartitionedRegionName;
            colocatedWith = CustomerPartitionedRegionName;
            isPartitionResolver = Boolean.FALSE;
            localMaxmemory = 0;
            redundancy = 0;
            totalNumBuckets = 11;
            PartitionAttributesFactory paf = new PartitionAttributesFactory();
            paf.setRedundantCopies(redundancy)
                .setLocalMaxMemory(localMaxmemory)
                .setTotalNumBuckets(totalNumBuckets).setColocatedWith(colocatedWith);
            if (isPartitionResolver) {
              paf.setPartitionResolver(
                  new CustomerIDPartitionResolver("CustomerIDPartitionResolver"));
            }
            PartitionAttributes prAttr = paf.create();
            AttributesFactory attr = new AttributesFactory();
            attr.setPartitionAttributes(prAttr);
            assertNotNull(basicGetCache());
            try {
              basicGetCache().createRegion(regionName, attr.create());
              fail("It should have failed with Exception: Colocated regions "
                  + "should have accessors at the same node");
            } catch (Exception Expected) {
              Expected.printStackTrace();
              LogWriterUtils.getLogWriter().info("Expected Message : " + Expected.getMessage());
              assertTrue(Expected.getMessage()
                  .startsWith("Colocated regions should have accessors at the same node"));
            }
          }
        });
    ex.remove();
  }

  @Test
  public void testColocatedPRWithAccessorOnDifferentNode2() throws Throwable {

    createCacheInAllVms();

    dataStore1.invoke(new CacheSerializableRunnable("testColocatedPRWithAccessorOnDifferentNode2") {
      @Override
      public void run2() {
        String partitionedRegionName = CustomerPartitionedRegionName;
        colocatedWith = null;
        isPartitionResolver = Boolean.FALSE;
        redundancy = 0;
        localMaxmemory = 0;
        totalNumBuckets = 11;
        PartitionAttributesFactory paf = new PartitionAttributesFactory();
        paf.setRedundantCopies(redundancy).setLocalMaxMemory(localMaxmemory)
            .setTotalNumBuckets(totalNumBuckets).setColocatedWith(colocatedWith);
        if (isPartitionResolver) {
          paf.setPartitionResolver(new CustomerIDPartitionResolver("CustomerIDPartitionResolver"));
        }
        PartitionAttributes prAttr = paf.create();
        AttributesFactory attr = new AttributesFactory();
        attr.setPartitionAttributes(prAttr);
        assertNotNull(basicGetCache());
        Region pr = basicGetCache().createRegion(partitionedRegionName, attr.create());
        assertNotNull(pr);
        LogWriterUtils.getLogWriter().info("Partitioned Region " + partitionedRegionName
            + " created Successfully :" + pr);
      }
    });

    // add expected exception string
    final IgnoredException ex = IgnoredException.addIgnoredException(
        "Colocated regions should have accessors at the same node", dataStore1);
    dataStore1
        .invoke(new CacheSerializableRunnable("Colocated PR with accessor on different nodes") {
          @Override
          public void run2() {
            regionName = OrderPartitionedRegionName;
            colocatedWith = CustomerPartitionedRegionName;
            isPartitionResolver = Boolean.FALSE;
            redundancy = 0;
            localMaxmemory = 50;
            totalNumBuckets = 11;
            PartitionAttributesFactory paf = new PartitionAttributesFactory();
            paf.setRedundantCopies(redundancy)
                .setLocalMaxMemory(localMaxmemory)
                .setTotalNumBuckets(totalNumBuckets).setColocatedWith(colocatedWith);
            if (isPartitionResolver) {
              paf.setPartitionResolver(
                  new CustomerIDPartitionResolver("CustomerIDPartitionResolver"));
            }
            PartitionAttributes prAttr = paf.create();
            AttributesFactory attr = new AttributesFactory();
            attr.setPartitionAttributes(prAttr);
            assertNotNull(basicGetCache());
            try {
              basicGetCache().createRegion(regionName, attr.create());
              fail("It should have failed with Exception: Colocated regions "
                  + "should have accessors at the same node");
            } catch (Exception Expected) {
              LogWriterUtils.getLogWriter().info("Expected Message : " + Expected.getMessage());
              assertTrue(Expected.getMessage()
                  .startsWith("Colocated regions should have accessors at the same node"));
            }
          }
        });
    ex.remove();
  }

  @Test
  public void testColocatedPRWithPROnDifferentNode1() throws Throwable {

    createCacheInAllVms();

