geode-core/src/distributedTest/java/org/apache/geode/internal/cache/PartitionListenerDUnitTest.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 org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;

 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;

 import org.junit.Test;

 import org.apache.geode.cache.AttributesFactory;
 import org.apache.geode.cache.Cache;
 import org.apache.geode.cache.PartitionAttributes;
 import org.apache.geode.cache.PartitionAttributesFactory;
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.control.RebalanceOperation;
 import org.apache.geode.cache.partition.PartitionListenerAdapter;
 import org.apache.geode.distributed.DistributedMember;
 import org.apache.geode.test.dunit.Host;
 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.cache.internal.JUnit4CacheTestCase;

 @SuppressWarnings({"serial", "rawtypes", "deprecation", "unchecked"})

 public class PartitionListenerDUnitTest extends JUnit4CacheTestCase {

   public PartitionListenerDUnitTest() {
     super();
   }

   @Test
   public void testAfterBucketRemovedCreated() throws Throwable {
     Host host = Host.getHost(0);
     VM vm0 = host.getVM(0);
     VM vm1 = host.getVM(1);
     VM vm2 = host.getVM(2);
     VM vm3 = host.getVM(3);

     // Create the PR in 2 JVMs
     String regionName = getName() + "_region";
     createPR(vm1, regionName, false);
     createPR(vm2, regionName, false);

     // Create the data using an accessor
     createPR(vm0, regionName, true);
     createData(vm0, 0, 1000, "A", regionName);

     // Assert that afterPrimary is invoked for every primary created on the vm
     List<Integer> vm1PrimariesCreated = getPrimariesCreated(vm1, regionName);
     List<Integer> vm2PrimariesCreated = getPrimariesCreated(vm2, regionName);
     List<Integer> vm1ActualPrimaries = getPrimariesOn(vm1, regionName);
     List<Integer> vm2ActualPrimaries = getPrimariesOn(vm2, regionName);

     vm1PrimariesCreated.removeAll(vm1ActualPrimaries);
     vm2PrimariesCreated.removeAll(vm2ActualPrimaries);

     assertThat(vm1PrimariesCreated).isEmpty();
     assertThat(vm2PrimariesCreated).isEmpty();

     // Create the PR in a third JVM and rebalance
     createPR(vm3, regionName, false);
     rebalance(vm3);

     // Verify listener invocations
     // Assert afterRegionCreate is invoked on every VM.
     assertEquals(regionName, getRegionNameFromListener(vm0, regionName));
     assertEquals(regionName, getRegionNameFromListener(vm1, regionName));
     assertEquals(regionName, getRegionNameFromListener(vm2, regionName));
     assertEquals(regionName, getRegionNameFromListener(vm3, regionName));

     // Get all buckets and keys removed from VM1 and VM2
     Map<Integer, List<Integer>> allBucketsAndKeysRemoved = new HashMap<>();
     allBucketsAndKeysRemoved.putAll(getBucketsAndKeysRemoved(vm1, regionName));
     allBucketsAndKeysRemoved.putAll(getBucketsAndKeysRemoved(vm2, regionName));

     // Get all buckets and keys added to VM3
     Map<Integer, List<Integer>> vm3BucketsAndKeysAdded = getBucketsAndKeysAdded(vm3, regionName);

     // Verify that they are equal
     assertEquals(allBucketsAndKeysRemoved, vm3BucketsAndKeysAdded);

     // Verify afterPrimary is invoked after rebalance.
     List<Integer> vm3PrimariesCreated = getPrimariesCreated(vm3, regionName);
     List<Integer> vm3ActualPrimaries = getPrimariesOn(vm3, regionName);

     vm3ActualPrimaries.removeAll(vm3PrimariesCreated);
     assertThat(vm3ActualPrimaries).isEmpty();
   }

   @Test
   public void testAfterSecondaryIsCalledAfterLosingPrimary() throws Throwable {
     Host host = Host.getHost(0);
     VM vm0 = host.getVM(0);
     VM vm1 = host.getVM(1);
     VM vm2 = host.getVM(2);
     VM vm3 = host.getVM(3);

     // Create the PR in 2 JVMs
     String regionName = getName() + "_region";
     createPR(vm1, regionName, false);
     createPR(vm2, regionName, false);

     // Create the data using an accessor
     createPR(vm0, regionName, true);
     createData(vm0, 0, 1000, "A", regionName);

     // record the bucket ids for each vm
     List<Integer> vm1ActualPrimaries = getPrimariesOn(vm1, regionName);
     List<Integer> vm2ActualPrimaries = getPrimariesOn(vm2, regionName);

     // Create the PR in a third JVM and rebalance
     createPR(vm3, regionName, false);
     rebalance(vm3);

     // Verify listener invocations
     List<Integer> afterSecondaryCalledForVM1 = getAfterSecondaryCallbackBucketIds(vm1, regionName);
     List<Integer> afterSecondaryCalledForVM2 = getAfterSecondaryCallbackBucketIds(vm2, regionName);

     // Eliminate the duplicate, prevent afterSecondary being called multiple times on the same
     // bucket
     Set<Integer> afterSecondaryCalledForVM1Set = new HashSet<>(afterSecondaryCalledForVM1);
     afterSecondaryCalledForVM1.removeAll(afterSecondaryCalledForVM1Set);
     assertTrue(
         "afterSecondary invoked more than once for bucket "
             + (afterSecondaryCalledForVM1.isEmpty() ? " " : afterSecondaryCalledForVM1.get(0)),
         afterSecondaryCalledForVM1.isEmpty());

     Set<Integer> afterSecondaryCalledForVM2Set = new HashSet<>(afterSecondaryCalledForVM2);
     afterSecondaryCalledForVM2.removeAll(afterSecondaryCalledForVM2Set);
     assertTrue(
         "afterSecondary invoked more than once for bucket "
             + (afterSecondaryCalledForVM2.isEmpty() ? " " : afterSecondaryCalledForVM2.get(0)),
         afterSecondaryCalledForVM2.isEmpty());

     List<Integer> newVm1ActualPrimaries = getPrimariesOn(vm1, regionName);
     List<Integer> newVM2ActualPrimaries = getPrimariesOn(vm2, regionName);

     // calculate and verify expected afterSecondary calls
     List<Integer> bucketsNoLongerPrimaryInVM1 = new ArrayList(vm1ActualPrimaries);
     bucketsNoLongerPrimaryInVM1.removeAll(newVm1ActualPrimaries);
     // GEODE-2785: it is possible a secondary bucket becomes primary during moving bucket stage,
     // and it then become secondary during primary selection stage. This may cause additional
     // afterSecondary callback being invoked.
     assertTrue(afterSecondaryCalledForVM1Set.containsAll(bucketsNoLongerPrimaryInVM1));

     List<Integer> bucketsNoLongerPrimaryInVM2 = new ArrayList(vm2ActualPrimaries);
     bucketsNoLongerPrimaryInVM2.removeAll(newVM2ActualPrimaries);
     assertTrue(afterSecondaryCalledForVM2Set.containsAll(bucketsNoLongerPrimaryInVM2));
   }

   protected DistributedMember createPR(VM vm, final String regionName, final boolean isAccessor)
       throws Throwable {
     SerializableCallable createPrRegion = new SerializableCallable("createRegion") {

       @Override
       public Object call() {
         Cache cache = getCache();
         AttributesFactory attr = new AttributesFactory();
         PartitionAttributesFactory paf = new PartitionAttributesFactory();
         paf.setRedundantCopies(1);
         if (isAccessor) {
           paf.setLocalMaxMemory(0);
         }
         paf.addPartitionListener(new TestPartitionListener());
         PartitionAttributes prAttr = paf.create();
         attr.setPartitionAttributes(prAttr);
         cache.createRegion(regionName, attr.create());
         return cache.getDistributedSystem().getDistributedMember();
       }
     };
     return (DistributedMember) vm.invoke(createPrRegion);
   }

   protected void createData(VM vm, final int startKey, final int endKey, final String value,
       final String regionName) {
     SerializableRunnable createData = new SerializableRunnable("createData") {

       @Override
       public void run() {
         Cache cache = getCache();
         Region region = cache.getRegion(regionName);

         for (int i = startKey; i < endKey; i++) {
           region.put(i, value);
         }
       }
     };
     vm.invoke(createData);
   }

   protected List<Integer> getPrimariesOn(VM vm, final String regionName) {
     SerializableCallable getPrimariesOn = new SerializableCallable("getPrimariesOn") {

       @Override
       public Object call() {
         Cache cache = getCache();
         Region region = cache.getRegion(regionName);
         return new ArrayList<>(
             ((PartitionedRegion) region).getDataStore().getAllLocalPrimaryBucketIds());
       }
     };
     return (List<Integer>) vm.invoke(getPrimariesOn);
   }

   protected List<Integer> getPrimariesCreated(VM vm, final String regionName) {
     SerializableCallable getPrimariesCreated = new SerializableCallable("getPrimariesCreated") {

       @Override
       public Object call() {
         Cache cache = getCache();
         Region region = cache.getRegion(regionName);
         TestPartitionListener listener = (TestPartitionListener) region.getAttributes()
             .getPartitionAttributes().getPartitionListeners()[0];
         return listener.getPrimariesCreated();
       }
     };
     return (List<Integer>) vm.invoke(getPrimariesCreated);
   }

   protected String getRegionNameFromListener(VM vm, final String regionName) {
     SerializableCallable getRegionName = new SerializableCallable("getRegionName") {

       @Override
       public Object call() {
         Cache cache = getCache();
         Region region = cache.getRegion(regionName);
         TestPartitionListener listener = (TestPartitionListener) region.getAttributes()
             .getPartitionAttributes().getPartitionListeners()[0];
         return listener.getRegionName();
       }
     };
     return (String) vm.invoke(getRegionName);
   }

   protected Map<Integer, List<Integer>> getBucketsAndKeysRemoved(VM vm, final String regionName) {
     SerializableCallable getBucketsAndKeysRemoved =
         new SerializableCallable("getBucketsAndKeysRemoved") {

           @Override
           public Object call() {
             Cache cache = getCache();
             Region region = cache.getRegion(regionName);
             TestPartitionListener listener = (TestPartitionListener) region.getAttributes()
                 .getPartitionAttributes().getPartitionListeners()[0];
             return listener.getBucketsAndKeysRemoved();
           }
         };
     return (Map<Integer, List<Integer>>) vm.invoke(getBucketsAndKeysRemoved);
   }

   protected Map<Integer, List<Integer>> getBucketsAndKeysAdded(VM vm, final String regionName) {
     SerializableCallable getBucketsAndKeysAdded =
         new SerializableCallable("getBucketsAndKeysAdded") {

           @Override
           public Object call() {
             Cache cache = getCache();
             Region region = cache.getRegion(regionName);
             TestPartitionListener listener = (TestPartitionListener) region.getAttributes()
                 .getPartitionAttributes().getPartitionListeners()[0];
             return listener.getBucketsAndKeysAdded();
           }
         };
     return (Map<Integer, List<Integer>>) vm.invoke(getBucketsAndKeysAdded);
   }

   protected List<Integer> getAfterSecondaryCallbackBucketIds(VM vm, final String regionName) {
     SerializableCallable getAfterSecondaryCallbackBucketIds =
         new SerializableCallable("getAfterSecondaryCallbackBucketIds") {

           @Override
           public Object call() {
             Cache cache = getCache();
             Region region = cache.getRegion(regionName);
             TestPartitionListener listener = (TestPartitionListener) region.getAttributes()
                 .getPartitionAttributes().getPartitionListeners()[0];
             return listener.getAfterSecondaryCallbackBucketIds();
           }
         };
     return (List<Integer>) vm.invoke(getAfterSecondaryCallbackBucketIds);
   }

   protected void rebalance(VM vm) {
     vm.invoke(new SerializableCallable() {

       @Override
       public Object call() throws Exception {
         RebalanceOperation rebalance =
             getCache().getResourceManager().createRebalanceFactory().start();
         rebalance.getResults();
         return null;
       }
     });
   }

   protected static class TestPartitionListener extends PartitionListenerAdapter {

     private String regionName;

     private final List<Integer> primariesCreated;

     private final List<Integer> afterSecondaryCalled;

     private final Map<Integer, List<Integer>> bucketsAndKeysRemoved;

     private final Map<Integer, List<Integer>> bucketsAndKeysAdded;

     public TestPartitionListener() {
       primariesCreated = new ArrayList<>();
       afterSecondaryCalled = new ArrayList<>();
       bucketsAndKeysRemoved = new HashMap<>();
       bucketsAndKeysAdded = new HashMap<>();
     }

     public Map<Integer, List<Integer>> getBucketsAndKeysRemoved() {
       return bucketsAndKeysRemoved;
     }

     public Map<Integer, List<Integer>> getBucketsAndKeysAdded() {
       return bucketsAndKeysAdded;
     }

     public List<Integer> getPrimariesCreated() {
       return primariesCreated;
     }

     public String getRegionName() {
       return regionName;
     }

     @Override
     public void afterRegionCreate(Region<?, ?> region) {
       regionName = region.getName();
     }

     @Override
     public void afterPrimary(int bucketId) {
       primariesCreated.add(bucketId);
     }

     @Override
     public void afterSecondary(int bucketId) {
       afterSecondaryCalled.add(bucketId);
     }

     public List<Integer> getAfterSecondaryCallbackBucketIds() {
       return afterSecondaryCalled;
     }

     @Override
     public void afterBucketRemoved(int bucketId, Iterable<?> keys) {
       Collection<Integer> keysCol = (Collection) keys;
       // If the keys collection is not empty, create a serializable list to hold
       // them and add them to the keys removed.
       if (!keysCol.isEmpty()) {
         List<Integer> keysList = new ArrayList<>();
         for (Integer key : keysCol) {
           keysList.add(key);
         }
         Collections.sort(keysList);
         bucketsAndKeysRemoved.put(bucketId, keysList);
       }
     }

     @Override
     public void afterBucketCreated(int bucketId, Iterable<?> keys) {
       Collection<Integer> keysCol = (Collection) keys;
       // If the keys collection is not empty, create a serializable list to hold
       // them and add them to the keys added.
       if (!keysCol.isEmpty()) {
         List<Integer> keysList = new ArrayList<>();
         for (Integer key : keysCol) {
           keysList.add(key);
         }
         Collections.sort(keysList);
         bucketsAndKeysAdded.put(bucketId, keysList);
       }
     }
   }
 }
	/*
	* 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 org.assertj.core.api.Assertions.assertThat;
	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertTrue;

	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.Collections;
	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;

	import org.junit.Test;

	import org.apache.geode.cache.AttributesFactory;
	import org.apache.geode.cache.Cache;
	import org.apache.geode.cache.PartitionAttributes;
	import org.apache.geode.cache.PartitionAttributesFactory;
	import org.apache.geode.cache.Region;
	import org.apache.geode.cache.control.RebalanceOperation;
	import org.apache.geode.cache.partition.PartitionListenerAdapter;
	import org.apache.geode.distributed.DistributedMember;
	import org.apache.geode.test.dunit.Host;
	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.cache.internal.JUnit4CacheTestCase;

	@SuppressWarnings({"serial", "rawtypes", "deprecation", "unchecked"})

	public class PartitionListenerDUnitTest extends JUnit4CacheTestCase {

	public PartitionListenerDUnitTest() {
	super();
	}

	@Test
	public void testAfterBucketRemovedCreated() throws Throwable {
	Host host = Host.getHost(0);
	VM vm0 = host.getVM(0);
	VM vm1 = host.getVM(1);
	VM vm2 = host.getVM(2);
	VM vm3 = host.getVM(3);

	// Create the PR in 2 JVMs
	String regionName = getName() + "_region";
	createPR(vm1, regionName, false);
	createPR(vm2, regionName, false);

	// Create the data using an accessor
	createPR(vm0, regionName, true);
	createData(vm0, 0, 1000, "A", regionName);

	// Assert that afterPrimary is invoked for every primary created on the vm
	List<Integer> vm1PrimariesCreated = getPrimariesCreated(vm1, regionName);
	List<Integer> vm2PrimariesCreated = getPrimariesCreated(vm2, regionName);
	List<Integer> vm1ActualPrimaries = getPrimariesOn(vm1, regionName);
	List<Integer> vm2ActualPrimaries = getPrimariesOn(vm2, regionName);

	vm1PrimariesCreated.removeAll(vm1ActualPrimaries);
	vm2PrimariesCreated.removeAll(vm2ActualPrimaries);

	assertThat(vm1PrimariesCreated).isEmpty();
	assertThat(vm2PrimariesCreated).isEmpty();

	// Create the PR in a third JVM and rebalance
	createPR(vm3, regionName, false);
	rebalance(vm3);

	// Verify listener invocations
	// Assert afterRegionCreate is invoked on every VM.
	assertEquals(regionName, getRegionNameFromListener(vm0, regionName));
	assertEquals(regionName, getRegionNameFromListener(vm1, regionName));
	assertEquals(regionName, getRegionNameFromListener(vm2, regionName));
	assertEquals(regionName, getRegionNameFromListener(vm3, regionName));

	// Get all buckets and keys removed from VM1 and VM2
	Map<Integer, List<Integer>> allBucketsAndKeysRemoved = new HashMap<>();
	allBucketsAndKeysRemoved.putAll(getBucketsAndKeysRemoved(vm1, regionName));
	allBucketsAndKeysRemoved.putAll(getBucketsAndKeysRemoved(vm2, regionName));

	// Get all buckets and keys added to VM3
	Map<Integer, List<Integer>> vm3BucketsAndKeysAdded = getBucketsAndKeysAdded(vm3, regionName);

	// Verify that they are equal
	assertEquals(allBucketsAndKeysRemoved, vm3BucketsAndKeysAdded);

	// Verify afterPrimary is invoked after rebalance.
	List<Integer> vm3PrimariesCreated = getPrimariesCreated(vm3, regionName);
	List<Integer> vm3ActualPrimaries = getPrimariesOn(vm3, regionName);

	vm3ActualPrimaries.removeAll(vm3PrimariesCreated);
	assertThat(vm3ActualPrimaries).isEmpty();
	}

	@Test
	public void testAfterSecondaryIsCalledAfterLosingPrimary() throws Throwable {
	Host host = Host.getHost(0);
	VM vm0 = host.getVM(0);
	VM vm1 = host.getVM(1);
	VM vm2 = host.getVM(2);
	VM vm3 = host.getVM(3);

	// Create the PR in 2 JVMs
	String regionName = getName() + "_region";
	createPR(vm1, regionName, false);
	createPR(vm2, regionName, false);

	// Create the data using an accessor
	createPR(vm0, regionName, true);
	createData(vm0, 0, 1000, "A", regionName);

	// record the bucket ids for each vm
	List<Integer> vm1ActualPrimaries = getPrimariesOn(vm1, regionName);
	List<Integer> vm2ActualPrimaries = getPrimariesOn(vm2, regionName);

	// Create the PR in a third JVM and rebalance
	createPR(vm3, regionName, false);
	rebalance(vm3);

	// Verify listener invocations
	List<Integer> afterSecondaryCalledForVM1 = getAfterSecondaryCallbackBucketIds(vm1, regionName);
	List<Integer> afterSecondaryCalledForVM2 = getAfterSecondaryCallbackBucketIds(vm2, regionName);

	// Eliminate the duplicate, prevent afterSecondary being called multiple times on the same
	// bucket
	Set<Integer> afterSecondaryCalledForVM1Set = new HashSet<>(afterSecondaryCalledForVM1);
	afterSecondaryCalledForVM1.removeAll(afterSecondaryCalledForVM1Set);
	assertTrue(
	"afterSecondary invoked more than once for bucket "
	+ (afterSecondaryCalledForVM1.isEmpty() ? " " : afterSecondaryCalledForVM1.get(0)),
	afterSecondaryCalledForVM1.isEmpty());

	Set<Integer> afterSecondaryCalledForVM2Set = new HashSet<>(afterSecondaryCalledForVM2);
	afterSecondaryCalledForVM2.removeAll(afterSecondaryCalledForVM2Set);
	assertTrue(
	"afterSecondary invoked more than once for bucket "
	+ (afterSecondaryCalledForVM2.isEmpty() ? " " : afterSecondaryCalledForVM2.get(0)),
	afterSecondaryCalledForVM2.isEmpty());

	List<Integer> newVm1ActualPrimaries = getPrimariesOn(vm1, regionName);
	List<Integer> newVM2ActualPrimaries = getPrimariesOn(vm2, regionName);

	// calculate and verify expected afterSecondary calls
	List<Integer> bucketsNoLongerPrimaryInVM1 = new ArrayList(vm1ActualPrimaries);
	bucketsNoLongerPrimaryInVM1.removeAll(newVm1ActualPrimaries);
	// GEODE-2785: it is possible a secondary bucket becomes primary during moving bucket stage,
	// and it then become secondary during primary selection stage. This may cause additional
	// afterSecondary callback being invoked.
	assertTrue(afterSecondaryCalledForVM1Set.containsAll(bucketsNoLongerPrimaryInVM1));

	List<Integer> bucketsNoLongerPrimaryInVM2 = new ArrayList(vm2ActualPrimaries);
	bucketsNoLongerPrimaryInVM2.removeAll(newVM2ActualPrimaries);
	assertTrue(afterSecondaryCalledForVM2Set.containsAll(bucketsNoLongerPrimaryInVM2));
	}

	protected DistributedMember createPR(VM vm, final String regionName, final boolean isAccessor)
	throws Throwable {
	SerializableCallable createPrRegion = new SerializableCallable("createRegion") {

	@Override
	public Object call() {
	Cache cache = getCache();
	AttributesFactory attr = new AttributesFactory();
	PartitionAttributesFactory paf = new PartitionAttributesFactory();
	paf.setRedundantCopies(1);
	if (isAccessor) {
	paf.setLocalMaxMemory(0);
	}
	paf.addPartitionListener(new TestPartitionListener());
	PartitionAttributes prAttr = paf.create();
	attr.setPartitionAttributes(prAttr);
	cache.createRegion(regionName, attr.create());
	return cache.getDistributedSystem().getDistributedMember();
	}
	};
	return (DistributedMember) vm.invoke(createPrRegion);
	}

	protected void createData(VM vm, final int startKey, final int endKey, final String value,
	final String regionName) {
	SerializableRunnable createData = new SerializableRunnable("createData") {

	@Override
	public void run() {
	Cache cache = getCache();
	Region region = cache.getRegion(regionName);

	for (int i = startKey; i < endKey; i++) {
	region.put(i, value);
	}
	}
	};
	vm.invoke(createData);
	}

	protected List<Integer> getPrimariesOn(VM vm, final String regionName) {
	SerializableCallable getPrimariesOn = new SerializableCallable("getPrimariesOn") {

	@Override
	public Object call() {
	Cache cache = getCache();
	Region region = cache.getRegion(regionName);
	return new ArrayList<>(
	((PartitionedRegion) region).getDataStore().getAllLocalPrimaryBucketIds());
	}
	};
	return (List<Integer>) vm.invoke(getPrimariesOn);
	}

	protected List<Integer> getPrimariesCreated(VM vm, final String regionName) {
	SerializableCallable getPrimariesCreated = new SerializableCallable("getPrimariesCreated") {

	@Override
	public Object call() {
	Cache cache = getCache();
	Region region = cache.getRegion(regionName);
	TestPartitionListener listener = (TestPartitionListener) region.getAttributes()
	.getPartitionAttributes().getPartitionListeners()[0];
	return listener.getPrimariesCreated();
	}
	};
	return (List<Integer>) vm.invoke(getPrimariesCreated);
	}

	protected String getRegionNameFromListener(VM vm, final String regionName) {
	SerializableCallable getRegionName = new SerializableCallable("getRegionName") {

	@Override
	public Object call() {
	Cache cache = getCache();
	Region region = cache.getRegion(regionName);
	TestPartitionListener listener = (TestPartitionListener) region.getAttributes()
	.getPartitionAttributes().getPartitionListeners()[0];
	return listener.getRegionName();
	}
	};
	return (String) vm.invoke(getRegionName);
	}

	protected Map<Integer, List<Integer>> getBucketsAndKeysRemoved(VM vm, final String regionName) {
	SerializableCallable getBucketsAndKeysRemoved =
	new SerializableCallable("getBucketsAndKeysRemoved") {

	@Override
	public Object call() {
	Cache cache = getCache();
	Region region = cache.getRegion(regionName);
	TestPartitionListener listener = (TestPartitionListener) region.getAttributes()
	.getPartitionAttributes().getPartitionListeners()[0];
	return listener.getBucketsAndKeysRemoved();
	}
	};
	return (Map<Integer, List<Integer>>) vm.invoke(getBucketsAndKeysRemoved);
	}

	protected Map<Integer, List<Integer>> getBucketsAndKeysAdded(VM vm, final String regionName) {
	SerializableCallable getBucketsAndKeysAdded =
	new SerializableCallable("getBucketsAndKeysAdded") {

	@Override
	public Object call() {
	Cache cache = getCache();
	Region region = cache.getRegion(regionName);
	TestPartitionListener listener = (TestPartitionListener) region.getAttributes()
	.getPartitionAttributes().getPartitionListeners()[0];
	return listener.getBucketsAndKeysAdded();
	}
	};
	return (Map<Integer, List<Integer>>) vm.invoke(getBucketsAndKeysAdded);
	}

	protected List<Integer> getAfterSecondaryCallbackBucketIds(VM vm, final String regionName) {
	SerializableCallable getAfterSecondaryCallbackBucketIds =
	new SerializableCallable("getAfterSecondaryCallbackBucketIds") {

	@Override
	public Object call() {
	Cache cache = getCache();
	Region region = cache.getRegion(regionName);
	TestPartitionListener listener = (TestPartitionListener) region.getAttributes()
	.getPartitionAttributes().getPartitionListeners()[0];
	return listener.getAfterSecondaryCallbackBucketIds();
	}
	};
	return (List<Integer>) vm.invoke(getAfterSecondaryCallbackBucketIds);
	}

	protected void rebalance(VM vm) {
	vm.invoke(new SerializableCallable() {

	@Override
	public Object call() throws Exception {
	RebalanceOperation rebalance =
	getCache().getResourceManager().createRebalanceFactory().start();
	rebalance.getResults();
	return null;
	}
	});
	}

	protected static class TestPartitionListener extends PartitionListenerAdapter {

	private String regionName;

	private final List<Integer> primariesCreated;

	private final List<Integer> afterSecondaryCalled;

	private final Map<Integer, List<Integer>> bucketsAndKeysRemoved;

	private final Map<Integer, List<Integer>> bucketsAndKeysAdded;

	public TestPartitionListener() {
	primariesCreated = new ArrayList<>();
	afterSecondaryCalled = new ArrayList<>();
	bucketsAndKeysRemoved = new HashMap<>();
	bucketsAndKeysAdded = new HashMap<>();
	}

	public Map<Integer, List<Integer>> getBucketsAndKeysRemoved() {
	return bucketsAndKeysRemoved;
	}

	public Map<Integer, List<Integer>> getBucketsAndKeysAdded() {
	return bucketsAndKeysAdded;
	}

	public List<Integer> getPrimariesCreated() {
	return primariesCreated;
	}

	public String getRegionName() {
	return regionName;
	}

	@Override
	public void afterRegionCreate(Region<?, ?> region) {
	regionName = region.getName();
	}

	@Override
	public void afterPrimary(int bucketId) {
	primariesCreated.add(bucketId);
	}

	@Override
	public void afterSecondary(int bucketId) {
	afterSecondaryCalled.add(bucketId);
	}

	public List<Integer> getAfterSecondaryCallbackBucketIds() {
	return afterSecondaryCalled;
	}

	@Override
	public void afterBucketRemoved(int bucketId, Iterable<?> keys) {
	Collection<Integer> keysCol = (Collection) keys;
	// If the keys collection is not empty, create a serializable list to hold
	// them and add them to the keys removed.
	if (!keysCol.isEmpty()) {
	List<Integer> keysList = new ArrayList<>();
	for (Integer key : keysCol) {
	keysList.add(key);
	}
	Collections.sort(keysList);
	bucketsAndKeysRemoved.put(bucketId, keysList);
	}
	}

	@Override
	public void afterBucketCreated(int bucketId, Iterable<?> keys) {
	Collection<Integer> keysCol = (Collection) keys;
	// If the keys collection is not empty, create a serializable list to hold
	// them and add them to the keys added.
	if (!keysCol.isEmpty()) {
	List<Integer> keysList = new ArrayList<>();
	for (Integer key : keysCol) {
	keysList.add(key);
	}
	Collections.sort(keysList);
	bucketsAndKeysAdded.put(bucketId, keysList);
	}
	}
	}
	}