geode-core/src/distributedTest/java/org/apache/geode/internal/cache/PartitionedRegionStatsDUnitTest.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.MINUTES;
 import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
 import static org.apache.geode.test.dunit.Host.getHost;
 import static org.assertj.core.api.Assertions.assertThat;

 import java.util.concurrent.TimeoutException;

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

 import org.apache.geode.Statistics;
 import org.apache.geode.cache.Cache;
 import org.apache.geode.cache.PartitionAttributesFactory;
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionFactory;
 import org.apache.geode.cache.RegionShortcut;
 import org.apache.geode.cache.control.RebalanceOperation;
 import org.apache.geode.test.dunit.VM;
 import org.apache.geode.test.dunit.cache.CacheTestCase;


 public class PartitionedRegionStatsDUnitTest extends CacheTestCase {

   private static final int CNT = 10;
   private static final int PUT = 1;
   private static final int GET = 2;
   private static final int CONTAINS_KEY = 3;
   private static final int CONTAINS_VALUE_FOR_KEY = 4;
   private static final int INVALIDATE = 5;
   private static final int DESTROY = 6;
   private static final int CREATE = 7;
   private static final int GET_ENTRY = 8;

   private static final int DEFAULT_NUM_BUCKETS = 113;
   public static final String STAT_CONFIGURED_REDUNDANT_COPIES = "configuredRedundantCopies";
   public static final String STAT_ACTUAL_REDUNDANT_COPIES = "actualRedundantCopies";

   private VM vm0;
   private VM vm1;
   private VM vm2;
   private VM vm3;

   private String regionName;

   @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);

     regionName = "region";
   }

   @Test
   public void testClose() throws Exception {
     // Create PRs on 3 VMs and accessors on 1 VM
     vm0.invoke(() -> createPartitionedRegion(regionName, 200, 1, 5));
     vm1.invoke(() -> createPartitionedRegion(regionName, 200, 1, 5));
     vm2.invoke(() -> createPartitionedRegion(regionName, 200, 1, 5));
     vm3.invoke(() -> createPartitionedRegion(regionName, 0, 1, 5));

     // Do Region operations.
     vm0.invoke(() -> doOpsOnRegion(regionName, PUT));
     vm0.invoke(() -> doOpsOnRegion(regionName, GET));
     vm0.invoke(() -> doOpsOnRegion(regionName, GET_ENTRY));
     vm0.invoke(() -> doOpsOnRegion(regionName, CONTAINS_KEY));
     vm0.invoke(() -> doOpsOnRegion(regionName, CONTAINS_VALUE_FOR_KEY));
     vm0.invoke(() -> doOpsOnRegion(regionName, INVALIDATE));
     vm0.invoke(() -> doOpsOnRegion(regionName, DESTROY));

     vm0.invoke(() -> validatePartitionedRegionOpsStats(regionName));

     vm0.invoke(() -> validateRedundantCopiesStats(regionName));
   }

   /**
    * Ok, first problem, GC'd tombstone is counted as an entry To test - modifying a tombstone -
    * modifying and doing tombstone GC?
    */
   @Test
   public void testDataStoreEntryCountWithRebalance() throws Exception {
     vm0.invoke(() -> createPartitionedRegionWithRebalance(regionName, 0));

     vm0.invoke(() -> {
       Cache cache = getCache();
       Region<Integer, Integer> region = cache.getRegion(regionName);
       region.put(0, 0);
       region.put(1, 1);
       region.put(113, 113);
       region.put(114, 114);
       region.destroy(0);
       region.destroy(1);
     });

     vm1.invoke(() -> createPartitionedRegionWithRebalance(regionName, 0));

     vm0.invoke(() -> validateEntryCount(regionName, 1));
     vm1.invoke(() -> validateEntryCount(regionName, 1));
   }

   @Test
   public void testDataStoreEntryCount2WithRebalance() throws Exception {
     vm0.invoke(() -> createPartitionedRegionWithRebalance(regionName, 0));

     vm0.invoke(() -> {
       Cache cache = getCache();
       Region<Integer, Integer> region = cache.getRegion(regionName);
       for (int i = 0; i <= 5; i++) {
         region.put(i, i);
       }
     });

     vm1.invoke(() -> createPartitionedRegionWithRebalance(regionName, 0));

     vm0.invoke(() -> validateEntryCount(regionName, 3));
     vm1.invoke(() -> validateEntryCount(regionName, 3));
   }

   /**
    * Test to make sure the datastore entry count is accurate.
    *
    * Ok, first problem, GC'd tombstone is counted as an entry To test - modifying a tombstone -
    * modifying and doing tombstone GC?
    */
   @Test
   public void testDataStoreEntryCount() throws Exception {
     vm0.invoke(() -> createPartitionedRegionWithRedundantCopies(regionName));
     vm1.invoke(() -> createPartitionedRegionWithRedundantCopies(regionName));

     vm0.invoke(() -> putDataInRegion(regionName));

     vm0.invoke(() -> validateEntryCount(regionName, 4));
     vm1.invoke(() -> validateEntryCount(regionName, 4));

     // Do a destroy
     vm0.invoke(() -> {
       Cache cache = getCache();
       Region<Integer, Integer> region = cache.getRegion(regionName);
       region.destroy(0);
     });

     // We expect the tombstone won't be recorded as part of the entry count
     vm0.invoke(() -> validateEntryCount(regionName, 3));
     vm1.invoke(() -> validateEntryCount(regionName, 3));

     // Destroy and modify a tombstone
     vm0.invoke(() -> {
       Cache cache = getCache();
       Region<Integer, Integer> region = cache.getRegion(regionName);
       region.destroy(113);
       region.put(113, 113);
     });

     vm0.invoke(() -> validateEntryCount(regionName, 3));
     vm1.invoke(() -> validateEntryCount(regionName, 3));

     // After GII (which might include the tombstone), a new members should still see only 2 live
     // entries.
     vm2.invoke(() -> createPartitionedRegionWithRedundantCopies(regionName));

     // Wait for redundancy to be restored. Once it is the entry count should be 2
     vm2.invoke(() -> {
       Cache cache = getCache();
       PartitionedRegion region = (PartitionedRegion) cache.getRegion(regionName);
       PartitionedRegionStats prStats = region.getPrStats();
       await()
           .untilAsserted(() -> assertThat(prStats.getLowRedundancyBucketCount()).isEqualTo(0));
     });

     vm2.invoke(() -> validateEntryCount(regionName, 3));

     // A tombstone GC shouldn't affect the count.
     vm0.invoke(() -> {
       InternalCache cache = getCache();
       TombstoneService tombstoneService = cache.getTombstoneService();
       tombstoneService.forceBatchExpirationForTests(1);
     });

     vm0.invoke(() -> validateEntryCount(regionName, 3));
     vm1.invoke(() -> validateEntryCount(regionName, 3));
     vm2.invoke(() -> validateEntryCount(regionName, 3));
   }

   @Test
   public void testTotalNumBuckets() {
     vm0.invoke(() -> createPartitionedRegionWithRedundantCopies(regionName));
     vm1.invoke(() -> createPartitionedRegionWithRedundantCopies(regionName));

     vm0.invoke(() -> putDataInRegion(regionName));
     vm1.invoke(() -> putDataInRegion(regionName));

     vm0.invoke(() -> validateTotalNumBucketsCount(regionName, DEFAULT_NUM_BUCKETS));
     vm1.invoke(() -> validateTotalNumBucketsCount(regionName, DEFAULT_NUM_BUCKETS));
   }

   private void putDataInRegion(final String regionName) {
     Cache cache = getCache();
     Region<Integer, Integer> region = cache.getRegion(regionName);
     region.put(0, 0);
     region.put(1, 1);
     region.put(113, 113);
     region.put(226, 114);
   }

   private void createPartitionedRegionWithRedundantCopies(final String regionName) {
     PartitionAttributesFactory paf = new PartitionAttributesFactory();
     paf.setRedundantCopies(2);

     RegionFactory regionFactory = getCache().createRegionFactory(RegionShortcut.PARTITION);
     regionFactory.setPartitionAttributes(paf.create());

     regionFactory.create(regionName);
   }

   private void createPartitionedRegion(final String regionName, final int localMaxMemory,
       final int redundancy, int totalNumBuckets) {
     PartitionAttributesFactory paf = new PartitionAttributesFactory();
     paf.setLocalMaxMemory(localMaxMemory);
     paf.setRedundantCopies(redundancy);
     paf.setTotalNumBuckets(totalNumBuckets);

     RegionFactory regionFactory = getCache().createRegionFactory(RegionShortcut.PARTITION);
     regionFactory.setPartitionAttributes(paf.create());

     regionFactory.create(regionName);
   }

   private void createPartitionedRegionWithRebalance(final String regionName, int redundancy)
       throws InterruptedException, TimeoutException {
     PartitionAttributesFactory paf = new PartitionAttributesFactory();
     paf.setRedundantCopies(redundancy);

     RegionFactory regionFactory = getCache().createRegionFactory(RegionShortcut.PARTITION);
     regionFactory.setPartitionAttributes(paf.create());

     regionFactory.create(regionName);

     RebalanceOperation op = getCache().getResourceManager().createRebalanceFactory().start();

     op.getResults(2, MINUTES);
   }

   private void validateEntryCount(final String regionName, final int expectedCount) {
     Cache cache = getCache();
     PartitionedRegion region = (PartitionedRegion) cache.getRegion(regionName);
     PartitionedRegionStats stats = region.getPrStats();
     CachePerfStats cachePerfStats = region.getCachePerfStats();

     assertThat(stats.getDataStoreEntryCount()).isEqualTo(expectedCount);
     long actualCount = cachePerfStats.stats.getLong(CachePerfStats.entryCountId);
     assertThat(actualCount).isEqualTo(expectedCount);
   }

   private void validateTotalNumBucketsCount(final String regionName, final int expectedCount) {
     Cache cache = getCache();
     PartitionedRegion region = (PartitionedRegion) cache.getRegion(regionName);
     PartitionedRegionStats stats = region.getPrStats();

     assertThat(stats.getTotalNumBuckets()).isEqualTo(expectedCount);
   }

   private void doOpsOnRegion(final String regionName, final int opType) {
     doOpsOnRegion(getCache().getRegion(regionName), opType);
   }

   private void doOpsOnRegion(final Region<Integer, Integer> region, final int opType) {
     switch (opType) {
       case PUT:
         for (int k = 0; k < CNT; k++) {
           region.put(k, k);
         }
         break;
       case GET:
         for (int k = 0; k < CNT; k++) {
           region.get(k);
         }
         break;
       case CONTAINS_KEY:
         for (int k = 0; k < CNT; k++) {
           region.containsKey(k);
         }
         break;
       case CONTAINS_VALUE_FOR_KEY:
         for (int k = 0; k < CNT; k++) {
           region.containsValueForKey(k);
         }
         break;
       case INVALIDATE:
         for (int k = 0; k < CNT; k++) {
           region.invalidate(k);
         }
         break;
       case DESTROY:
         for (int k = 0; k < CNT; k++) {
           region.destroy(k);
         }
         break;
       case CREATE:
         for (int k = 0; k < CNT; k++) {
           region.create(k, k);
         }
         break;
       case GET_ENTRY:
         for (int k = 0; k < CNT; k++) {
           region.getEntry(k);
         }
     }
   }

   private void validatePartitionedRegionOpsStats(final String regionName) {
     Cache cache = getCache();
     PartitionedRegion partitionedRegion = (PartitionedRegion) cache.getRegion(regionName);

     assertThat(partitionedRegion).isNotNull();

     Statistics stats = partitionedRegion.getPrStats().getStats();

     int putsCompleted = stats.get("putsCompleted").intValue();
     int getsCompleted = stats.get("getsCompleted").intValue();
     int getEntryCompleted = stats.get("getEntryCompleted").intValue();
     int createsCompleted = stats.get("createsCompleted").intValue();
     int containsKeyCompleted = stats.get("containsKeyCompleted").intValue();
     int containsValueForKeyCompleted = stats.get("containsValueForKeyCompleted").intValue();
     int invalidatesCompleted = stats.get("invalidatesCompleted").intValue();
     int destroysCompleted = stats.get("destroysCompleted").intValue();

     assertThat(putsCompleted).isEqualTo(CNT);
     assertThat(getsCompleted).isEqualTo(CNT);
     assertThat(getEntryCompleted).isEqualTo(CNT);
     assertThat(createsCompleted).isEqualTo(0);
     assertThat(containsKeyCompleted).isEqualTo(CNT);
     assertThat(containsValueForKeyCompleted).isEqualTo(CNT);
     assertThat(invalidatesCompleted).isEqualTo(CNT);
     assertThat(destroysCompleted).isEqualTo(CNT);

     // TODO: Need to validate that bucket stats add up....
     // partitionedRegion.getDataStore().getCachePerfStats().getGets(); // etc...
   }

   private void validateRedundantCopiesStats(final String regionName) {
     Cache cache = getCache();
     PartitionedRegion partitionedRegion = (PartitionedRegion) cache.getRegion(regionName);

     assertThat(partitionedRegion).isNotNull();

     Statistics stats = partitionedRegion.getPrStats().getStats();
     int configuredRedundantCopies = stats.get(STAT_CONFIGURED_REDUNDANT_COPIES).intValue();
     int actualRedundantCopies = stats.get(STAT_ACTUAL_REDUNDANT_COPIES).intValue();

     assertThat(configuredRedundantCopies).isEqualTo(1);
     assertThat(actualRedundantCopies).isEqualTo(1);
   }
 }
	/*
	* 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.MINUTES;
	import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
	import static org.apache.geode.test.dunit.Host.getHost;
	import static org.assertj.core.api.Assertions.assertThat;

	import java.util.concurrent.TimeoutException;

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

	import org.apache.geode.Statistics;
	import org.apache.geode.cache.Cache;
	import org.apache.geode.cache.PartitionAttributesFactory;
	import org.apache.geode.cache.Region;
	import org.apache.geode.cache.RegionFactory;
	import org.apache.geode.cache.RegionShortcut;
	import org.apache.geode.cache.control.RebalanceOperation;
	import org.apache.geode.test.dunit.VM;
	import org.apache.geode.test.dunit.cache.CacheTestCase;


	public class PartitionedRegionStatsDUnitTest extends CacheTestCase {

	private static final int CNT = 10;
	private static final int PUT = 1;
	private static final int GET = 2;
	private static final int CONTAINS_KEY = 3;
	private static final int CONTAINS_VALUE_FOR_KEY = 4;
	private static final int INVALIDATE = 5;
	private static final int DESTROY = 6;
	private static final int CREATE = 7;
	private static final int GET_ENTRY = 8;

	private static final int DEFAULT_NUM_BUCKETS = 113;
	public static final String STAT_CONFIGURED_REDUNDANT_COPIES = "configuredRedundantCopies";
	public static final String STAT_ACTUAL_REDUNDANT_COPIES = "actualRedundantCopies";

	private VM vm0;
	private VM vm1;
	private VM vm2;
	private VM vm3;

	private String regionName;

	@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);

	regionName = "region";
	}

	@Test
	public void testClose() throws Exception {
	// Create PRs on 3 VMs and accessors on 1 VM
	vm0.invoke(() -> createPartitionedRegion(regionName, 200, 1, 5));
	vm1.invoke(() -> createPartitionedRegion(regionName, 200, 1, 5));
	vm2.invoke(() -> createPartitionedRegion(regionName, 200, 1, 5));
	vm3.invoke(() -> createPartitionedRegion(regionName, 0, 1, 5));

	// Do Region operations.
	vm0.invoke(() -> doOpsOnRegion(regionName, PUT));
	vm0.invoke(() -> doOpsOnRegion(regionName, GET));
	vm0.invoke(() -> doOpsOnRegion(regionName, GET_ENTRY));
	vm0.invoke(() -> doOpsOnRegion(regionName, CONTAINS_KEY));
	vm0.invoke(() -> doOpsOnRegion(regionName, CONTAINS_VALUE_FOR_KEY));
	vm0.invoke(() -> doOpsOnRegion(regionName, INVALIDATE));
	vm0.invoke(() -> doOpsOnRegion(regionName, DESTROY));

	vm0.invoke(() -> validatePartitionedRegionOpsStats(regionName));

	vm0.invoke(() -> validateRedundantCopiesStats(regionName));
	}

	/**
	* Ok, first problem, GC'd tombstone is counted as an entry To test - modifying a tombstone -
	* modifying and doing tombstone GC?
	*/
	@Test
	public void testDataStoreEntryCountWithRebalance() throws Exception {
	vm0.invoke(() -> createPartitionedRegionWithRebalance(regionName, 0));

	vm0.invoke(() -> {
	Cache cache = getCache();
	Region<Integer, Integer> region = cache.getRegion(regionName);
	region.put(0, 0);
	region.put(1, 1);
	region.put(113, 113);
	region.put(114, 114);
	region.destroy(0);
	region.destroy(1);
	});

	vm1.invoke(() -> createPartitionedRegionWithRebalance(regionName, 0));

	vm0.invoke(() -> validateEntryCount(regionName, 1));
	vm1.invoke(() -> validateEntryCount(regionName, 1));
	}

	@Test
	public void testDataStoreEntryCount2WithRebalance() throws Exception {
	vm0.invoke(() -> createPartitionedRegionWithRebalance(regionName, 0));

	vm0.invoke(() -> {
	Cache cache = getCache();
	Region<Integer, Integer> region = cache.getRegion(regionName);
	for (int i = 0; i <= 5; i++) {
	region.put(i, i);
	}
	});

	vm1.invoke(() -> createPartitionedRegionWithRebalance(regionName, 0));

	vm0.invoke(() -> validateEntryCount(regionName, 3));
	vm1.invoke(() -> validateEntryCount(regionName, 3));
	}

	/**
	* Test to make sure the datastore entry count is accurate.
	*
	* Ok, first problem, GC'd tombstone is counted as an entry To test - modifying a tombstone -
	* modifying and doing tombstone GC?
	*/
	@Test
	public void testDataStoreEntryCount() throws Exception {
	vm0.invoke(() -> createPartitionedRegionWithRedundantCopies(regionName));
	vm1.invoke(() -> createPartitionedRegionWithRedundantCopies(regionName));

	vm0.invoke(() -> putDataInRegion(regionName));

	vm0.invoke(() -> validateEntryCount(regionName, 4));
	vm1.invoke(() -> validateEntryCount(regionName, 4));

	// Do a destroy
	vm0.invoke(() -> {
	Cache cache = getCache();
	Region<Integer, Integer> region = cache.getRegion(regionName);
	region.destroy(0);
	});

	// We expect the tombstone won't be recorded as part of the entry count
	vm0.invoke(() -> validateEntryCount(regionName, 3));
	vm1.invoke(() -> validateEntryCount(regionName, 3));

	// Destroy and modify a tombstone
	vm0.invoke(() -> {
	Cache cache = getCache();
	Region<Integer, Integer> region = cache.getRegion(regionName);
	region.destroy(113);
	region.put(113, 113);
	});

	vm0.invoke(() -> validateEntryCount(regionName, 3));
	vm1.invoke(() -> validateEntryCount(regionName, 3));

	// After GII (which might include the tombstone), a new members should still see only 2 live
	// entries.
	vm2.invoke(() -> createPartitionedRegionWithRedundantCopies(regionName));

	// Wait for redundancy to be restored. Once it is the entry count should be 2
	vm2.invoke(() -> {
	Cache cache = getCache();
	PartitionedRegion region = (PartitionedRegion) cache.getRegion(regionName);
	PartitionedRegionStats prStats = region.getPrStats();
	await()
	.untilAsserted(() -> assertThat(prStats.getLowRedundancyBucketCount()).isEqualTo(0));
	});

	vm2.invoke(() -> validateEntryCount(regionName, 3));

	// A tombstone GC shouldn't affect the count.
	vm0.invoke(() -> {
	InternalCache cache = getCache();
	TombstoneService tombstoneService = cache.getTombstoneService();
	tombstoneService.forceBatchExpirationForTests(1);
	});

	vm0.invoke(() -> validateEntryCount(regionName, 3));
	vm1.invoke(() -> validateEntryCount(regionName, 3));
	vm2.invoke(() -> validateEntryCount(regionName, 3));
	}

	@Test
	public void testTotalNumBuckets() {
	vm0.invoke(() -> createPartitionedRegionWithRedundantCopies(regionName));
	vm1.invoke(() -> createPartitionedRegionWithRedundantCopies(regionName));

	vm0.invoke(() -> putDataInRegion(regionName));
	vm1.invoke(() -> putDataInRegion(regionName));

	vm0.invoke(() -> validateTotalNumBucketsCount(regionName, DEFAULT_NUM_BUCKETS));
	vm1.invoke(() -> validateTotalNumBucketsCount(regionName, DEFAULT_NUM_BUCKETS));
	}

	private void putDataInRegion(final String regionName) {
	Cache cache = getCache();
	Region<Integer, Integer> region = cache.getRegion(regionName);
	region.put(0, 0);
	region.put(1, 1);
	region.put(113, 113);
	region.put(226, 114);
	}

	private void createPartitionedRegionWithRedundantCopies(final String regionName) {
	PartitionAttributesFactory paf = new PartitionAttributesFactory();
	paf.setRedundantCopies(2);

	RegionFactory regionFactory = getCache().createRegionFactory(RegionShortcut.PARTITION);
	regionFactory.setPartitionAttributes(paf.create());

	regionFactory.create(regionName);
	}

	private void createPartitionedRegion(final String regionName, final int localMaxMemory,
	final int redundancy, int totalNumBuckets) {
	PartitionAttributesFactory paf = new PartitionAttributesFactory();
	paf.setLocalMaxMemory(localMaxMemory);
	paf.setRedundantCopies(redundancy);
	paf.setTotalNumBuckets(totalNumBuckets);

	RegionFactory regionFactory = getCache().createRegionFactory(RegionShortcut.PARTITION);
	regionFactory.setPartitionAttributes(paf.create());

	regionFactory.create(regionName);
	}

	private void createPartitionedRegionWithRebalance(final String regionName, int redundancy)
	throws InterruptedException, TimeoutException {
	PartitionAttributesFactory paf = new PartitionAttributesFactory();
	paf.setRedundantCopies(redundancy);

	RegionFactory regionFactory = getCache().createRegionFactory(RegionShortcut.PARTITION);
	regionFactory.setPartitionAttributes(paf.create());

	regionFactory.create(regionName);

	RebalanceOperation op = getCache().getResourceManager().createRebalanceFactory().start();

	op.getResults(2, MINUTES);
	}

	private void validateEntryCount(final String regionName, final int expectedCount) {
	Cache cache = getCache();
	PartitionedRegion region = (PartitionedRegion) cache.getRegion(regionName);
	PartitionedRegionStats stats = region.getPrStats();
	CachePerfStats cachePerfStats = region.getCachePerfStats();

	assertThat(stats.getDataStoreEntryCount()).isEqualTo(expectedCount);
	long actualCount = cachePerfStats.stats.getLong(CachePerfStats.entryCountId);
	assertThat(actualCount).isEqualTo(expectedCount);
	}

	private void validateTotalNumBucketsCount(final String regionName, final int expectedCount) {
	Cache cache = getCache();
	PartitionedRegion region = (PartitionedRegion) cache.getRegion(regionName);
	PartitionedRegionStats stats = region.getPrStats();

	assertThat(stats.getTotalNumBuckets()).isEqualTo(expectedCount);
	}

	private void doOpsOnRegion(final String regionName, final int opType) {
	doOpsOnRegion(getCache().getRegion(regionName), opType);
	}

	private void doOpsOnRegion(final Region<Integer, Integer> region, final int opType) {
	switch (opType) {
	case PUT:
	for (int k = 0; k < CNT; k++) {
	region.put(k, k);
	}
	break;
	case GET:
	for (int k = 0; k < CNT; k++) {
	region.get(k);
	}
	break;
	case CONTAINS_KEY:
	for (int k = 0; k < CNT; k++) {
	region.containsKey(k);
	}
	break;
	case CONTAINS_VALUE_FOR_KEY:
	for (int k = 0; k < CNT; k++) {
	region.containsValueForKey(k);
	}
	break;
	case INVALIDATE:
	for (int k = 0; k < CNT; k++) {
	region.invalidate(k);
	}
	break;
	case DESTROY:
	for (int k = 0; k < CNT; k++) {
	region.destroy(k);
	}
	break;
	case CREATE:
	for (int k = 0; k < CNT; k++) {
	region.create(k, k);
	}
	break;
	case GET_ENTRY:
	for (int k = 0; k < CNT; k++) {
	region.getEntry(k);
	}
	}
	}

	private void validatePartitionedRegionOpsStats(final String regionName) {
	Cache cache = getCache();
	PartitionedRegion partitionedRegion = (PartitionedRegion) cache.getRegion(regionName);

	assertThat(partitionedRegion).isNotNull();

	Statistics stats = partitionedRegion.getPrStats().getStats();

	int putsCompleted = stats.get("putsCompleted").intValue();
	int getsCompleted = stats.get("getsCompleted").intValue();
	int getEntryCompleted = stats.get("getEntryCompleted").intValue();
	int createsCompleted = stats.get("createsCompleted").intValue();
	int containsKeyCompleted = stats.get("containsKeyCompleted").intValue();
	int containsValueForKeyCompleted = stats.get("containsValueForKeyCompleted").intValue();
	int invalidatesCompleted = stats.get("invalidatesCompleted").intValue();
	int destroysCompleted = stats.get("destroysCompleted").intValue();

	assertThat(putsCompleted).isEqualTo(CNT);
	assertThat(getsCompleted).isEqualTo(CNT);
	assertThat(getEntryCompleted).isEqualTo(CNT);
	assertThat(createsCompleted).isEqualTo(0);
	assertThat(containsKeyCompleted).isEqualTo(CNT);
	assertThat(containsValueForKeyCompleted).isEqualTo(CNT);
	assertThat(invalidatesCompleted).isEqualTo(CNT);
	assertThat(destroysCompleted).isEqualTo(CNT);

	// TODO: Need to validate that bucket stats add up....
	// partitionedRegion.getDataStore().getCachePerfStats().getGets(); // etc...
	}

	private void validateRedundantCopiesStats(final String regionName) {
	Cache cache = getCache();
	PartitionedRegion partitionedRegion = (PartitionedRegion) cache.getRegion(regionName);

	assertThat(partitionedRegion).isNotNull();

	Statistics stats = partitionedRegion.getPrStats().getStats();
	int configuredRedundantCopies = stats.get(STAT_CONFIGURED_REDUNDANT_COPIES).intValue();
	int actualRedundantCopies = stats.get(STAT_ACTUAL_REDUNDANT_COPIES).intValue();

	assertThat(configuredRedundantCopies).isEqualTo(1);
	assertThat(actualRedundantCopies).isEqualTo(1);
	}
	}