geode-core/src/distributedTest/java/org/apache/geode/internal/cache/PartitionedRegionCreationDUnitTest.java - geode - Git at Google

 /*
  * 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.SECONDS;
 import static org.apache.geode.cache.RegionShortcut.PARTITION;
 import static org.apache.geode.internal.cache.PartitionedRegionHelper.PR_ROOT_REGION_NAME;
 import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
 import static org.apache.geode.test.dunit.Host.getHost;
 import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;

 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;

 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;

 import org.apache.geode.cache.Cache;
 import org.apache.geode.cache.DataPolicy;
 import org.apache.geode.cache.PartitionAttributesFactory;
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionAttributes;
 import org.apache.geode.cache.RegionFactory;
 import org.apache.geode.cache.RegionShortcut;
 import org.apache.geode.test.dunit.AsyncInvocation;
 import org.apache.geode.test.dunit.IgnoredException;
 import org.apache.geode.test.dunit.SerializableRunnableIF;
 import org.apache.geode.test.dunit.VM;
 import org.apache.geode.test.dunit.cache.CacheTestCase;
 import org.apache.geode.test.junit.rules.ExecutorServiceRule;

 /**
  * This test is to test and validate the partitioned region creation in multiple vm scenario. This
  * will verify the functionality under distributed scenario.
  */
 @SuppressWarnings("serial")
 public class PartitionedRegionCreationDUnitTest extends CacheTestCase {

   private VM vm0;
   private VM vm1;
   private VM vm2;
   private VM vm3;
   private String signalRegionName;
   private String partitionedRegionName;
   private int numberOfRegions;
   private String prNamePrefix;

   private transient List<AsyncInvocation<Void>> asyncInvocations;

   @Rule
   public ExecutorServiceRule executorServiceRule = new ExecutorServiceRule();

   @Before
   public void setUp() throws Exception {
     vm0 = getHost(0).getVM(0);
     vm1 = getHost(0).getVM(1);
     vm2 = getHost(0).getVM(2);
     vm3 = getHost(0).getVM(3);

     partitionedRegionName = getUniqueName() + "_PartitionedRegion";
     prNamePrefix = partitionedRegionName + "-";
     signalRegionName = getUniqueName() + "_Signal";
     numberOfRegions = 1;
     asyncInvocations = new ArrayList<>();
   }

   /**
    * This tests creates partition regions with scope = DISTRIBUTED_ACK and then validating thoes
    * partition regions
    */
   @Test
   public void testSequentialCreation() throws Exception {
     for (int i = 0; i < numberOfRegions; i++) {
       final int whichRegion = i;
       vm0.invoke(() -> createPartitionedRegion(prNamePrefix + whichRegion));
       vm1.invoke(() -> createPartitionedRegion(prNamePrefix + whichRegion));
       vm2.invoke(() -> createPartitionedRegion(prNamePrefix + whichRegion));
       vm3.invoke(() -> createPartitionedRegion(prNamePrefix + whichRegion));
     }

     // validating that regions are successfully created
     vm0.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
     vm1.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
     vm2.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
     vm3.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
   }

   /**
    * This test create regions with scope = DISTRIBUTED_NO_ACK and then validating these partition
    * regions
    */
   @Test
   public void testConcurrentCreation() throws Exception {
     invokeAsync(vm0, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
     invokeAsync(vm1, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
     invokeAsync(vm2, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
     invokeAsync(vm3, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));

     awaitAllAsyncInvocations();

     // validating that regions are successfully created
     vm0.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
     vm1.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
     vm2.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
     vm3.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
   }

   /**
    * This test create regions with scope = DISTRIBUTED_NO_ACK and then validating these partition
    * regions. Test specially added for SQL fabric testing since that always creates regions in
    * parallel.
    */
   @Test
   public void testBetterConcurrentCreation() throws Exception {
     numberOfRegions = 3;

     createReplicateRegion(signalRegionName);
     vm0.invoke(() -> createReplicateRegion(signalRegionName));
     vm1.invoke(() -> createReplicateRegion(signalRegionName));
     vm2.invoke(() -> createReplicateRegion(signalRegionName));
     vm3.invoke(() -> createReplicateRegion(signalRegionName));

     CompletableFuture<Void> createdAccessor = executorServiceRule.runAsync(
         () -> createPRAccessorConcurrently(signalRegionName, prNamePrefix, numberOfRegions));

     invokeAsync(vm0, () -> createPRConcurrently(signalRegionName, prNamePrefix, numberOfRegions));
     invokeAsync(vm1, () -> createPRConcurrently(signalRegionName, prNamePrefix, numberOfRegions));
     invokeAsync(vm2, () -> createPRConcurrently(signalRegionName, prNamePrefix, numberOfRegions));
     invokeAsync(vm3, () -> createPRConcurrently(signalRegionName, prNamePrefix, numberOfRegions));

     // do the put to signal start
     Region<String, String> signalRegion = getCache().getRegion(signalRegionName);
     signalRegion.put("start", "true");

     awaitAllAsyncInvocations();
     createdAccessor.get(60, SECONDS);

     // validating that regions are successfully created
     vm0.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
     vm1.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
     vm2.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
     vm3.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
     validatePR(prNamePrefix, numberOfRegions);
   }

   /**
    * Test whether partition region creation is prevented when an instance is created that has the
    * incorrect redundancy
    */
   @Test
   public void testPartitionedRegionRedundancyConflict() throws Exception {
     vm0.invoke(() -> {
       Cache cache = getCache();

       PartitionAttributesFactory partitionAttributesFactory = new PartitionAttributesFactory();
       partitionAttributesFactory.setRedundantCopies(0);

       RegionFactory regionFactory = cache.createRegionFactory(PARTITION);
       regionFactory.setPartitionAttributes(partitionAttributesFactory.create());

       Region region = regionFactory.create(partitionedRegionName);

       assertThat(region).isNotNull();
       assertThat(region.isDestroyed()).isFalse();
       assertThat(cache.getRegion(partitionedRegionName)).isNotNull();
     });

     for (int i = 1; i < 4; i++) {
       final int redundancy = i;

       vm1.invoke(() -> {
         Cache cache = getCache();

         PartitionAttributesFactory partitionAttributesFactory = new PartitionAttributesFactory();
         partitionAttributesFactory.setRedundantCopies(redundancy);

         RegionFactory regionFactory = cache.createRegionFactory(PARTITION);
         regionFactory.setPartitionAttributes(partitionAttributesFactory.create());

         try (IgnoredException ie = addIgnoredException(IllegalStateException.class)) {
           assertThatThrownBy(() -> regionFactory.create(partitionedRegionName))
               .isInstanceOf(IllegalStateException.class);
         }

         assertThat(cache.getRegion(partitionedRegionName)).isNull();
       });
     }
   }

   /**
    * This test creates partition region with scope = DISTRIBUTED_ACK and tests whether all the
    * attributes of partitioned region are properly initialized
    */
   @Test
   public void testPartitionRegionInitialization() throws Exception {
     invokeAsync(vm0, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
     invokeAsync(vm1, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
     invokeAsync(vm2, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
     invokeAsync(vm3, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));

     awaitAllAsyncInvocations();

     invokeAsync(vm0, this::validatePRInitialization);
     invokeAsync(vm1, this::validatePRInitialization);
     invokeAsync(vm2, this::validatePRInitialization);
     invokeAsync(vm3, this::validatePRInitialization);

     awaitAllAsyncInvocations();
   }

   @Test
   public void testPartitionRegionRegistration() throws Exception {
     invokeAsync(vm0, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
     invokeAsync(vm1, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
     invokeAsync(vm2, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
     invokeAsync(vm3, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));

     awaitAllAsyncInvocations();

     invokeAsync(vm0, () -> validatePRRegistration(prNamePrefix, numberOfRegions));
     invokeAsync(vm1, () -> validatePRRegistration(prNamePrefix, numberOfRegions));
     invokeAsync(vm2, () -> validatePRRegistration(prNamePrefix, numberOfRegions));
     invokeAsync(vm3, () -> validatePRRegistration(prNamePrefix, numberOfRegions));

     awaitAllAsyncInvocations();
   }

   /**
    * This tests persistence conflicts btw members of partition region
    */
   @Test
   public void testPartitionRegionPersistenceConflicts() throws Exception {
     VM dataStore0 = vm0;
     VM dataStore1 = vm1;
     VM accessor0 = vm2;
     VM accessor1 = vm3;

     addIgnoredException("IllegalStateException");

     accessor0.invoke(() -> createPersistentPR(partitionedRegionName, 0, false));
     accessor1
         .invoke(() -> createPersistentPRThrowsIfLocalMaxMemoryIsZero(partitionedRegionName, true));
     dataStore0.invoke(() -> createPersistentPR(partitionedRegionName, 100, true));
     dataStore1.invoke(
         () -> createPersistentPRThrowsIfDataPolicyConflict(partitionedRegionName, 100, false));
   }

   private void validatePRInitialization() {
     Cache cache = getCache();
     Region root = cache.getRegion(PR_ROOT_REGION_NAME);
     assertThat(root).isNotNull();

     RegionAttributes regionAttributes = root.getAttributes();
     assertThat(regionAttributes.getScope().isDistributedAck()).isTrue();
     assertThat(regionAttributes.getDataPolicy()).isEqualTo(DataPolicy.REPLICATE);
   }

   private void validatePRRegistration(final String prPrefixName, final int count) {
     InternalCache cache = getCache();
     Region root = PartitionedRegionHelper.getPRRoot(cache);
     assertThat(root).isNotNull();

     for (int i = 0; i < count; i++) {
       PartitionedRegion region = (PartitionedRegion) cache.getRegion(prPrefixName + i);
       assertThat(region).isNotNull();
       String name = region.getRegionIdentifier();
       PartitionRegionConfig partitionRegionConfig = (PartitionRegionConfig) root.get(name);
       assertThat(partitionRegionConfig).isNotNull();
     }
   }

   public void createPartitionedRegion(final String regionName) {
     Cache cache = getCache();
     RegionFactory regionFactory = cache.createRegionFactory(PARTITION);
     Region region = regionFactory.create(regionName);

     assertThat(region).isNotNull();
     assertThat(region.isDestroyed()).isFalse();
     assertThat(cache.getRegion(regionName)).isNotNull();
   }

   private void createPartitionedRegions(final String prPrefixName, final int count) {
     Cache cache = getCache();
     RegionFactory regionFactory = cache.createRegionFactory(PARTITION);

     for (int i = 0; i < count; i++) {
       Region region = regionFactory.create(prPrefixName + i);
       assertThat(region).isNotNull();
       assertThat(region.isDestroyed()).isFalse();
       assertThat(cache.getRegion(prPrefixName + i)).isNotNull();
     }
   }

   private void createPersistentPR(final String regionName, final int localMaxMemory,
       final boolean isPersistent) {
     Cache cache = getCache();

     RegionFactory regionFactory = cache.createRegionFactory(RegionShortcut.PARTITION);

     if (isPersistent) {
       regionFactory.setDataPolicy(DataPolicy.PERSISTENT_PARTITION);
     } else {
       regionFactory.setDataPolicy(DataPolicy.PARTITION);
     }

     PartitionAttributesFactory partitionAttributesFactory = new PartitionAttributesFactory();
     partitionAttributesFactory.setLocalMaxMemory(localMaxMemory);
     regionFactory.setPartitionAttributes(partitionAttributesFactory.create());

     Region region = regionFactory.create(regionName);

     assertThat(region).isNotNull();
     assertThat(region.isDestroyed()).isFalse();
     assertThat(cache.getRegion(regionName)).isNotNull();
   }

   private void createPersistentPRThrowsIfLocalMaxMemoryIsZero(final String regionName,
       final boolean isPersistent) {
     String message = "Persistence is not allowed when local-max-memory is zero.";
     createPersistentPRThrows(regionName, 0, isPersistent, message);
   }

   private void createPersistentPRThrowsIfDataPolicyConflict(final String regionName,
       final int localMaxMemory, final boolean isPersistent) {
     assertThat(localMaxMemory).isGreaterThan(0);
     String message = "DataPolicy for Datastore members should all be persistent or not.";
     createPersistentPRThrows(regionName, localMaxMemory, isPersistent, message);
   }

   private void createPersistentPRThrows(final String regionName, final int localMaxMemory,
       final boolean isPersistent, final String message) {
     Cache cache = getCache();

     RegionFactory regionFactory = cache.createRegionFactory(RegionShortcut.PARTITION);

     if (isPersistent) {
       regionFactory.setDataPolicy(DataPolicy.PERSISTENT_PARTITION);
     } else {
       regionFactory.setDataPolicy(DataPolicy.PARTITION);
     }

     PartitionAttributesFactory partitionAttributesFactory = new PartitionAttributesFactory();
     partitionAttributesFactory.setLocalMaxMemory(localMaxMemory);
     regionFactory.setPartitionAttributes(partitionAttributesFactory.create());

     assertThatThrownBy(() -> regionFactory.create(regionName))
         .isInstanceOf(IllegalStateException.class).hasMessageContaining(message);
   }

   private void validatePR(final String regionName, final int count) {
     Cache cache = getCache();
     for (int i = 0; i < count; i++) {
       assertThat(cache.getRegion(regionName + i)).isNotNull();
     }
   }

   private void createReplicateRegion(final String regionName) {
     Cache cache = getCache();
     RegionFactory regionFactory = cache.createRegionFactory(RegionShortcut.REPLICATE);
     regionFactory.create(regionName);
   }

   private void createPRConcurrently(final String regionName, final String prPrefixName,
       final int count) {
     createPRConcurrently(regionName, prPrefixName, count, false);
   }

   private void createPRAccessorConcurrently(final String regionName, final String prPrefixName,
       final int count) {
     createPRConcurrently(regionName, prPrefixName, count, true);
   }

   private void createPRConcurrently(final String regionName, final String prPrefixName,
       final int count, final boolean accessor) {
     Cache cache = getCache();

     PartitionAttributesFactory partitionAttributesFactory = new PartitionAttributesFactory();
     partitionAttributesFactory.setRedundantCopies(2);
     if (accessor) {
       partitionAttributesFactory.setLocalMaxMemory(0);
     }

     RegionFactory regionFactory = cache.createRegionFactory(RegionShortcut.PARTITION);
     regionFactory.setPartitionAttributes(partitionAttributesFactory.create());

     // wait for put
     Region region = cache.getRegion(regionName);

     await().until(() -> region.getEntry("start") != null);

     for (int i = 0; i < count; ++i) {
       Region partitionedRegion = regionFactory.create(prPrefixName + i);

       assertThat(partitionedRegion).isNotNull();
       assertThat(partitionedRegion.isDestroyed()).isFalse();
       assertThat(cache.getRegion(regionName)).isNotNull();
     }
   }

   private void invokeAsync(final VM vm, final SerializableRunnableIF runnable) {
     asyncInvocations.add(vm.invokeAsync(runnable));
   }

   private void awaitAllAsyncInvocations() throws ExecutionException, InterruptedException {
     for (AsyncInvocation<Void> async : asyncInvocations) {
       async.await();
     }
     asyncInvocations.clear();
   }
 }
	/*
	* 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.SECONDS;
	import static org.apache.geode.cache.RegionShortcut.PARTITION;
	import static org.apache.geode.internal.cache.PartitionedRegionHelper.PR_ROOT_REGION_NAME;
	import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
	import static org.apache.geode.test.dunit.Host.getHost;
	import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
	import static org.assertj.core.api.Assertions.assertThat;
	import static org.assertj.core.api.Assertions.assertThatThrownBy;

	import java.util.ArrayList;
	import java.util.List;
	import java.util.concurrent.CompletableFuture;
	import java.util.concurrent.ExecutionException;

	import org.junit.Before;
	import org.junit.Rule;
	import org.junit.Test;

	import org.apache.geode.cache.Cache;
	import org.apache.geode.cache.DataPolicy;
	import org.apache.geode.cache.PartitionAttributesFactory;
	import org.apache.geode.cache.Region;
	import org.apache.geode.cache.RegionAttributes;
	import org.apache.geode.cache.RegionFactory;
	import org.apache.geode.cache.RegionShortcut;
	import org.apache.geode.test.dunit.AsyncInvocation;
	import org.apache.geode.test.dunit.IgnoredException;
	import org.apache.geode.test.dunit.SerializableRunnableIF;
	import org.apache.geode.test.dunit.VM;
	import org.apache.geode.test.dunit.cache.CacheTestCase;
	import org.apache.geode.test.junit.rules.ExecutorServiceRule;

	/**
	* This test is to test and validate the partitioned region creation in multiple vm scenario. This
	* will verify the functionality under distributed scenario.
	*/
	@SuppressWarnings("serial")
	public class PartitionedRegionCreationDUnitTest extends CacheTestCase {

	private VM vm0;
	private VM vm1;
	private VM vm2;
	private VM vm3;
	private String signalRegionName;
	private String partitionedRegionName;
	private int numberOfRegions;
	private String prNamePrefix;

	private transient List<AsyncInvocation<Void>> asyncInvocations;

	@Rule
	public ExecutorServiceRule executorServiceRule = new ExecutorServiceRule();

	@Before
	public void setUp() throws Exception {
	vm0 = getHost(0).getVM(0);
	vm1 = getHost(0).getVM(1);
	vm2 = getHost(0).getVM(2);
	vm3 = getHost(0).getVM(3);

	partitionedRegionName = getUniqueName() + "_PartitionedRegion";
	prNamePrefix = partitionedRegionName + "-";
	signalRegionName = getUniqueName() + "_Signal";
	numberOfRegions = 1;
	asyncInvocations = new ArrayList<>();
	}

	/**
	* This tests creates partition regions with scope = DISTRIBUTED_ACK and then validating thoes
	* partition regions
	*/
	@Test
	public void testSequentialCreation() throws Exception {
	for (int i = 0; i < numberOfRegions; i++) {
	final int whichRegion = i;
	vm0.invoke(() -> createPartitionedRegion(prNamePrefix + whichRegion));
	vm1.invoke(() -> createPartitionedRegion(prNamePrefix + whichRegion));
	vm2.invoke(() -> createPartitionedRegion(prNamePrefix + whichRegion));
	vm3.invoke(() -> createPartitionedRegion(prNamePrefix + whichRegion));
	}

	// validating that regions are successfully created
	vm0.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
	vm1.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
	vm2.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
	vm3.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
	}

	/**
	* This test create regions with scope = DISTRIBUTED_NO_ACK and then validating these partition
	* regions
	*/
	@Test
	public void testConcurrentCreation() throws Exception {
	invokeAsync(vm0, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
	invokeAsync(vm1, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
	invokeAsync(vm2, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
	invokeAsync(vm3, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));

	awaitAllAsyncInvocations();

	// validating that regions are successfully created
	vm0.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
	vm1.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
	vm2.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
	vm3.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
	}

	/**
	* This test create regions with scope = DISTRIBUTED_NO_ACK and then validating these partition
	* regions. Test specially added for SQL fabric testing since that always creates regions in
	* parallel.
	*/
	@Test
	public void testBetterConcurrentCreation() throws Exception {
	numberOfRegions = 3;

	createReplicateRegion(signalRegionName);
	vm0.invoke(() -> createReplicateRegion(signalRegionName));
	vm1.invoke(() -> createReplicateRegion(signalRegionName));
	vm2.invoke(() -> createReplicateRegion(signalRegionName));
	vm3.invoke(() -> createReplicateRegion(signalRegionName));

	CompletableFuture<Void> createdAccessor = executorServiceRule.runAsync(
	() -> createPRAccessorConcurrently(signalRegionName, prNamePrefix, numberOfRegions));

	invokeAsync(vm0, () -> createPRConcurrently(signalRegionName, prNamePrefix, numberOfRegions));
	invokeAsync(vm1, () -> createPRConcurrently(signalRegionName, prNamePrefix, numberOfRegions));
	invokeAsync(vm2, () -> createPRConcurrently(signalRegionName, prNamePrefix, numberOfRegions));
	invokeAsync(vm3, () -> createPRConcurrently(signalRegionName, prNamePrefix, numberOfRegions));

	// do the put to signal start
	Region<String, String> signalRegion = getCache().getRegion(signalRegionName);
	signalRegion.put("start", "true");

	awaitAllAsyncInvocations();
	createdAccessor.get(60, SECONDS);

	// validating that regions are successfully created
	vm0.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
	vm1.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
	vm2.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
	vm3.invoke(() -> validatePR(prNamePrefix, numberOfRegions));
	validatePR(prNamePrefix, numberOfRegions);
	}

	/**
	* Test whether partition region creation is prevented when an instance is created that has the
	* incorrect redundancy
	*/
	@Test
	public void testPartitionedRegionRedundancyConflict() throws Exception {
	vm0.invoke(() -> {
	Cache cache = getCache();

	PartitionAttributesFactory partitionAttributesFactory = new PartitionAttributesFactory();
	partitionAttributesFactory.setRedundantCopies(0);

	RegionFactory regionFactory = cache.createRegionFactory(PARTITION);
	regionFactory.setPartitionAttributes(partitionAttributesFactory.create());

	Region region = regionFactory.create(partitionedRegionName);

	assertThat(region).isNotNull();
	assertThat(region.isDestroyed()).isFalse();
	assertThat(cache.getRegion(partitionedRegionName)).isNotNull();
	});

	for (int i = 1; i < 4; i++) {
	final int redundancy = i;

	vm1.invoke(() -> {
	Cache cache = getCache();

	PartitionAttributesFactory partitionAttributesFactory = new PartitionAttributesFactory();
	partitionAttributesFactory.setRedundantCopies(redundancy);

	RegionFactory regionFactory = cache.createRegionFactory(PARTITION);
	regionFactory.setPartitionAttributes(partitionAttributesFactory.create());

	try (IgnoredException ie = addIgnoredException(IllegalStateException.class)) {
	assertThatThrownBy(() -> regionFactory.create(partitionedRegionName))
	.isInstanceOf(IllegalStateException.class);
	}

	assertThat(cache.getRegion(partitionedRegionName)).isNull();
	});
	}
	}

	/**
	* This test creates partition region with scope = DISTRIBUTED_ACK and tests whether all the
	* attributes of partitioned region are properly initialized
	*/
	@Test
	public void testPartitionRegionInitialization() throws Exception {
	invokeAsync(vm0, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
	invokeAsync(vm1, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
	invokeAsync(vm2, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
	invokeAsync(vm3, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));

	awaitAllAsyncInvocations();

	invokeAsync(vm0, this::validatePRInitialization);
	invokeAsync(vm1, this::validatePRInitialization);
	invokeAsync(vm2, this::validatePRInitialization);
	invokeAsync(vm3, this::validatePRInitialization);

	awaitAllAsyncInvocations();
	}

	@Test
	public void testPartitionRegionRegistration() throws Exception {
	invokeAsync(vm0, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
	invokeAsync(vm1, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
	invokeAsync(vm2, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));
	invokeAsync(vm3, () -> createPartitionedRegions(prNamePrefix, numberOfRegions));

	awaitAllAsyncInvocations();

	invokeAsync(vm0, () -> validatePRRegistration(prNamePrefix, numberOfRegions));
	invokeAsync(vm1, () -> validatePRRegistration(prNamePrefix, numberOfRegions));
	invokeAsync(vm2, () -> validatePRRegistration(prNamePrefix, numberOfRegions));
	invokeAsync(vm3, () -> validatePRRegistration(prNamePrefix, numberOfRegions));

	awaitAllAsyncInvocations();
	}

	/**
	* This tests persistence conflicts btw members of partition region
	*/
	@Test
	public void testPartitionRegionPersistenceConflicts() throws Exception {
	VM dataStore0 = vm0;
	VM dataStore1 = vm1;
	VM accessor0 = vm2;
	VM accessor1 = vm3;

	addIgnoredException("IllegalStateException");

	accessor0.invoke(() -> createPersistentPR(partitionedRegionName, 0, false));
	accessor1
	.invoke(() -> createPersistentPRThrowsIfLocalMaxMemoryIsZero(partitionedRegionName, true));
	dataStore0.invoke(() -> createPersistentPR(partitionedRegionName, 100, true));
	dataStore1.invoke(
	() -> createPersistentPRThrowsIfDataPolicyConflict(partitionedRegionName, 100, false));
	}

	private void validatePRInitialization() {
	Cache cache = getCache();
	Region root = cache.getRegion(PR_ROOT_REGION_NAME);
	assertThat(root).isNotNull();

	RegionAttributes regionAttributes = root.getAttributes();
	assertThat(regionAttributes.getScope().isDistributedAck()).isTrue();
	assertThat(regionAttributes.getDataPolicy()).isEqualTo(DataPolicy.REPLICATE);
	}

	private void validatePRRegistration(final String prPrefixName, final int count) {
	InternalCache cache = getCache();
	Region root = PartitionedRegionHelper.getPRRoot(cache);
	assertThat(root).isNotNull();

	for (int i = 0; i < count; i++) {
	PartitionedRegion region = (PartitionedRegion) cache.getRegion(prPrefixName + i);
	assertThat(region).isNotNull();
	String name = region.getRegionIdentifier();
	PartitionRegionConfig partitionRegionConfig = (PartitionRegionConfig) root.get(name);
	assertThat(partitionRegionConfig).isNotNull();
	}
	}

	public void createPartitionedRegion(final String regionName) {
	Cache cache = getCache();
	RegionFactory regionFactory = cache.createRegionFactory(PARTITION);
	Region region = regionFactory.create(regionName);

	assertThat(region).isNotNull();
	assertThat(region.isDestroyed()).isFalse();
	assertThat(cache.getRegion(regionName)).isNotNull();
	}

	private void createPartitionedRegions(final String prPrefixName, final int count) {
	Cache cache = getCache();
	RegionFactory regionFactory = cache.createRegionFactory(PARTITION);

	for (int i = 0; i < count; i++) {
	Region region = regionFactory.create(prPrefixName + i);
	assertThat(region).isNotNull();
	assertThat(region.isDestroyed()).isFalse();
	assertThat(cache.getRegion(prPrefixName + i)).isNotNull();
	}
	}

	private void createPersistentPR(final String regionName, final int localMaxMemory,
	final boolean isPersistent) {
	Cache cache = getCache();

	RegionFactory regionFactory = cache.createRegionFactory(RegionShortcut.PARTITION);

	if (isPersistent) {
	regionFactory.setDataPolicy(DataPolicy.PERSISTENT_PARTITION);
	} else {
	regionFactory.setDataPolicy(DataPolicy.PARTITION);
	}

	PartitionAttributesFactory partitionAttributesFactory = new PartitionAttributesFactory();
	partitionAttributesFactory.setLocalMaxMemory(localMaxMemory);
	regionFactory.setPartitionAttributes(partitionAttributesFactory.create());

	Region region = regionFactory.create(regionName);

	assertThat(region).isNotNull();
	assertThat(region.isDestroyed()).isFalse();
	assertThat(cache.getRegion(regionName)).isNotNull();
	}

	private void createPersistentPRThrowsIfLocalMaxMemoryIsZero(final String regionName,
	final boolean isPersistent) {
	String message = "Persistence is not allowed when local-max-memory is zero.";
	createPersistentPRThrows(regionName, 0, isPersistent, message);
	}

	private void createPersistentPRThrowsIfDataPolicyConflict(final String regionName,
	final int localMaxMemory, final boolean isPersistent) {
	assertThat(localMaxMemory).isGreaterThan(0);
	String message = "DataPolicy for Datastore members should all be persistent or not.";
	createPersistentPRThrows(regionName, localMaxMemory, isPersistent, message);
	}

	private void createPersistentPRThrows(final String regionName, final int localMaxMemory,
	final boolean isPersistent, final String message) {
	Cache cache = getCache();

	RegionFactory regionFactory = cache.createRegionFactory(RegionShortcut.PARTITION);

	if (isPersistent) {
	regionFactory.setDataPolicy(DataPolicy.PERSISTENT_PARTITION);
	} else {
	regionFactory.setDataPolicy(DataPolicy.PARTITION);
	}

	PartitionAttributesFactory partitionAttributesFactory = new PartitionAttributesFactory();
	partitionAttributesFactory.setLocalMaxMemory(localMaxMemory);
	regionFactory.setPartitionAttributes(partitionAttributesFactory.create());

	assertThatThrownBy(() -> regionFactory.create(regionName))
	.isInstanceOf(IllegalStateException.class).hasMessageContaining(message);
	}

	private void validatePR(final String regionName, final int count) {
	Cache cache = getCache();
	for (int i = 0; i < count; i++) {
	assertThat(cache.getRegion(regionName + i)).isNotNull();
	}
	}

	private void createReplicateRegion(final String regionName) {
	Cache cache = getCache();
	RegionFactory regionFactory = cache.createRegionFactory(RegionShortcut.REPLICATE);
	regionFactory.create(regionName);
	}

	private void createPRConcurrently(final String regionName, final String prPrefixName,
	final int count) {
	createPRConcurrently(regionName, prPrefixName, count, false);
	}

	private void createPRAccessorConcurrently(final String regionName, final String prPrefixName,
	final int count) {
	createPRConcurrently(regionName, prPrefixName, count, true);
	}

	private void createPRConcurrently(final String regionName, final String prPrefixName,
	final int count, final boolean accessor) {
	Cache cache = getCache();

	PartitionAttributesFactory partitionAttributesFactory = new PartitionAttributesFactory();
	partitionAttributesFactory.setRedundantCopies(2);
	if (accessor) {
	partitionAttributesFactory.setLocalMaxMemory(0);
	}

	RegionFactory regionFactory = cache.createRegionFactory(RegionShortcut.PARTITION);
	regionFactory.setPartitionAttributes(partitionAttributesFactory.create());

	// wait for put
	Region region = cache.getRegion(regionName);

	await().until(() -> region.getEntry("start") != null);

	for (int i = 0; i < count; ++i) {
	Region partitionedRegion = regionFactory.create(prPrefixName + i);

	assertThat(partitionedRegion).isNotNull();
	assertThat(partitionedRegion.isDestroyed()).isFalse();
	assertThat(cache.getRegion(regionName)).isNotNull();
	}
	}

	private void invokeAsync(final VM vm, final SerializableRunnableIF runnable) {
	asyncInvocations.add(vm.invokeAsync(runnable));
	}

	private void awaitAllAsyncInvocations() throws ExecutionException, InterruptedException {
	for (AsyncInvocation<Void> async : asyncInvocations) {
	async.await();
	}
	asyncInvocations.clear();
	}
	}