geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jasonhu...@apache.org
Subject [18/37] incubator-geode git commit: GEODE-75: System property to Rebalance multiple regions in parallel
Date Thu, 23 Jul 2015 16:18:55 GMT
GEODE-75: System property to Rebalance multiple regions in parallel

Allow multiple regions to be rebalanced in parallel, by setting
gemfire.resource.manager.threads to something greater than 1.


Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/f7242d23
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/f7242d23
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/f7242d23

Branch: refs/heads/feature/GEODE-9
Commit: f7242d23c0d256b4d2e93e9ce325f8255959c9af
Parents: d993684
Author: Dan Smith <dsmith@pivotal.io>
Authored: Mon Jun 15 14:04:33 2015 -0700
Committer: Dan Smith <dsmith@pivotal.io>
Committed: Tue Jul 14 12:10:14 2015 -0700

----------------------------------------------------------------------
 .../cache/control/InternalResourceManager.java  |  15 +-
 .../cache/control/RebalanceOperationImpl.java   | 157 +++++++++----
 .../cache/control/RebalanceResultsImpl.java     |  16 +-
 .../control/RebalanceOperationDUnitTest.java    | 228 ++++++++++++++++++-
 4 files changed, 359 insertions(+), 57 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/f7242d23/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/control/InternalResourceManager.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/control/InternalResourceManager.java
b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/control/InternalResourceManager.java
index d144aaf..1479e28 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/control/InternalResourceManager.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/control/InternalResourceManager.java
@@ -20,6 +20,7 @@ import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.logging.log4j.Logger;
 
@@ -36,7 +37,6 @@ import com.gemstone.gemfire.distributed.internal.OverflowQueueWithDMStats;
 import com.gemstone.gemfire.distributed.internal.SerialQueuedExecutorWithDMStats;
 import com.gemstone.gemfire.internal.ClassPathLoader;
 import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
-import com.gemstone.gemfire.internal.cache.LocalRegion;
 import com.gemstone.gemfire.internal.cache.PartitionedRegion;
 import com.gemstone.gemfire.internal.cache.control.ResourceAdvisor.ResourceManagerProfile;
 import com.gemstone.gemfire.internal.cache.partitioned.LoadProbe;
@@ -57,6 +57,8 @@ import com.gemstone.gemfire.internal.logging.log4j.LocalizedMessage;
  */
 public class InternalResourceManager implements ResourceManager {
   private static final Logger logger = LogService.getLogger();
+
+  final int MAX_RESOURCE_MANAGER_EXE_THREADS = Integer.getInteger("gemfire.resource.manager.threads",
1);
   
   public enum ResourceType {
     HEAP_MEMORY(0x1), OFFHEAP_MEMORY(0x2), MEMORY(0x3), ALL(0xFFFFFFFF);
@@ -107,18 +109,23 @@ public class InternalResourceManager implements ResourceManager {
     
     // Create a new executor that other classes may use for handling resource
     // related tasks
-    final ThreadGroup thrdGrp = LoggingThreadGroup.createThreadGroup(
+    final ThreadGroup threadGroup = LoggingThreadGroup.createThreadGroup(
         "ResourceManagerThreadGroup", logger);
 
     ThreadFactory tf = new ThreadFactory() {
+      AtomicInteger ai = new AtomicInteger();
       @Override
       public Thread newThread(Runnable r) {
-        Thread thread = new Thread(thrdGrp, r, "ResourceManagerRecoveryThread");
+        int tId = ai.getAndIncrement();
+        Thread thread = new Thread(threadGroup, r,
+        "ResourceManagerRecoveryThread " + tId);
         thread.setDaemon(true);
         return thread;
       }
     };
-    this.scheduledExecutor = new ScheduledThreadPoolExecutor(1, tf);
+    int nThreads = MAX_RESOURCE_MANAGER_EXE_THREADS;
+
+    this.scheduledExecutor = new ScheduledThreadPoolExecutor(nThreads, tf);
 
     // Initialize the load probe
     try {

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/f7242d23/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/control/RebalanceOperationImpl.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/control/RebalanceOperationImpl.java
b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/control/RebalanceOperationImpl.java
index be4d824..75d1d17 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/control/RebalanceOperationImpl.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/control/RebalanceOperationImpl.java
@@ -8,11 +8,15 @@
 
 package com.gemstone.gemfire.internal.cache.control;
 
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -45,7 +49,8 @@ public class RebalanceOperationImpl implements RebalanceOperation {
   
   private final boolean simulation;
   private final GemFireCacheImpl cache;
-  private Future<RebalanceResults> future;
+  private List<Future<RebalanceResults>> futureList = new ArrayList<Future<RebalanceResults>>();
+  private int pendingTasks;
   private final AtomicBoolean cancelled = new AtomicBoolean();
   private final Object futureLock = new Object();
   private RegionFilter filter;
@@ -59,30 +64,15 @@ public class RebalanceOperationImpl implements RebalanceOperation {
     
   public void start() {
     final InternalResourceManager manager = this.cache.getResourceManager();
-    ScheduledExecutorService ex = manager.getExecutor();
     synchronized (this.futureLock) {
       manager.addInProgressRebalance(this);
-      future = ex.submit(new Callable<RebalanceResults>() {
-        public RebalanceResults call() {
-          SystemFailure.checkFailure();
-          cache.getCancelCriterion().checkCancelInProgress(null);
-          try {
-            return RebalanceOperationImpl.this.call();
-          }
-          catch (RuntimeException e) {
-            logger.debug("Unexpected exception in rebalancing: {}", e.getMessage(), e);
-            throw e;
-          } finally {
-            manager.removeInProgressRebalance(RebalanceOperationImpl.this);
-          }
-        }
-      });
+      this.scheduleRebalance();      
     }
   }
   
-  private RebalanceResults call() {
-    RebalanceResultsImpl results = new RebalanceResultsImpl();
+  private void scheduleRebalance() {
     ResourceManagerStats stats = cache.getResourceManager().getStats();
+    
     long start = stats.startRebalance();
     try {
     for(PartitionedRegion region: cache.getPartitionedRegions()) {
@@ -93,13 +83,12 @@ public class RebalanceOperationImpl implements RebalanceOperation {
         //Colocated regions will be rebalanced as part of rebalancing their leader
           if (region.getColocatedWith() == null && filter.include(region)) {
             
-            Set<PartitionRebalanceInfo> detailSet = null;
             if (region.isFixedPartitionedRegion()) {
               if (Boolean.getBoolean("gemfire.DISABLE_MOVE_PRIMARIES_ON_STARTUP")) {
                 PartitionedRegionRebalanceOp prOp = new PartitionedRegionRebalanceOp(
                     region, simulation, new CompositeDirector(false, false, false, true),
true, true, cancelled,
                     stats);
-                detailSet = prOp.execute();
+                this.futureList.add(submitRebalanceTask(prOp,start));
               } else {
                 continue;
               }
@@ -107,39 +96,100 @@ public class RebalanceOperationImpl implements RebalanceOperation {
               PartitionedRegionRebalanceOp prOp = new PartitionedRegionRebalanceOp(
                   region, simulation, new CompositeDirector(true, true, true, true), true,
true, cancelled,
                   stats);
-              detailSet = prOp.execute();
-            }
-            for (PartitionRebalanceInfo details : detailSet) {
-              results.addDetails(details);
-            }
+              this.futureList.add(submitRebalanceTask(prOp,start));
+            }            
           }
       } catch(RegionDestroyedException e) {
         //ignore, go on to the next region
       }
     }
     } finally {
-      stats.endRebalance(start);
+      if(pendingTasks == 0) {
+        //if we didn't submit any tasks, end the rebalance now.
+        stats.endRebalance(start);
+      }
     }
-    return results;
   }
   
-  private Future<RebalanceResults> getFuture() {
-    synchronized (this.futureLock) {
-      return this.future;
+  private Future<RebalanceResults> submitRebalanceTask(final PartitionedRegionRebalanceOp
rebalanceOp, final long rebalanceStartTime) {
+    final InternalResourceManager manager = this.cache.getResourceManager();
+    ScheduledExecutorService ex = manager.getExecutor();
+
+    synchronized(futureLock) {
+      //this update should happen inside this.futureLock 
+      pendingTasks++;
+
+      try {
+        Future<RebalanceResults> future = ex.submit(new Callable<RebalanceResults>()
{
+          public RebalanceResults call() {
+            try {
+              RebalanceResultsImpl results = new RebalanceResultsImpl();
+              SystemFailure.checkFailure();
+              cache.getCancelCriterion().checkCancelInProgress(null);
+
+              Set<PartitionRebalanceInfo> detailSet = null;
+
+              detailSet = rebalanceOp.execute();
+
+              for (PartitionRebalanceInfo details : detailSet) {
+                results.addDetails(details);
+              }
+              return results;
+            }
+            catch (RuntimeException e) {
+              logger.debug("Unexpected exception in rebalancing: {}", e.getMessage(), e);
+              throw e;
+            } finally {
+              synchronized (RebalanceOperationImpl.this.futureLock) {
+                pendingTasks--;
+                if(pendingTasks == 0) {//all threads done
+                  manager.removeInProgressRebalance(RebalanceOperationImpl.this);
+                  manager.getStats().endRebalance(rebalanceStartTime);
+                }
+              }
+            }
+          }
+        });
+
+        return future;
+      } catch(RejectedExecutionException e) {
+        cache.getCancelCriterion().checkCancelInProgress(null);
+        throw e;
+      }
+    }
+  }
+  
+  private List<Future<RebalanceResults>> getFutureList() {
+    synchronized(this.futureList) {
+      return this.futureList;
     }
   }
   
   public boolean cancel() {
     cancelled.set(true);
-    if(getFuture().cancel(false)) {
-      cache.getResourceManager().removeInProgressRebalance(this);
+    
+    synchronized (this.futureLock) {
+      for(Future<RebalanceResults> fr : getFutureList()) {
+        if(fr.cancel(false)) {
+          pendingTasks--;
+        }
+      }
+      if(pendingTasks == 0 ) {
+        cache.getResourceManager().removeInProgressRebalance(this);
+      }
     }
+    
     return true;
   }
 
   public RebalanceResults getResults() throws CancellationException, InterruptedException
{
+    RebalanceResultsImpl results = new RebalanceResultsImpl();
+    List<Future<RebalanceResults>> frlist =  getFutureList();
+    for(Future<RebalanceResults> fr : frlist) {
       try {
-        return getFuture().get();
+        RebalanceResults rr =  fr.get();
+        results.addDetails((RebalanceResultsImpl)rr);
+        
       } catch (ExecutionException e) {
         if(e.getCause() instanceof GemFireException) {
           throw (GemFireException) e.getCause();
@@ -149,29 +199,48 @@ public class RebalanceOperationImpl implements RebalanceOperation {
           throw new InternalGemFireError(e.getCause());
         }
       }
+    }
+    return results;
   }
 
   public RebalanceResults getResults(long timeout, TimeUnit unit)
       throws CancellationException, TimeoutException, InterruptedException {
-    try {
-      return getFuture().get(timeout, unit);
-    } catch (ExecutionException e) {
-      if(e.getCause() instanceof GemFireException) {
-        throw (GemFireException) e.getCause();
-      } else if(e.getCause() instanceof InternalGemFireError) {
-        throw (InternalGemFireError) e.getCause();
-      } else {
-        throw new InternalGemFireError(e.getCause());
+    long endTime = unit.toNanos(timeout) + System.nanoTime();
+    
+    RebalanceResultsImpl results = new RebalanceResultsImpl();
+    List<Future<RebalanceResults>> frlist =  getFutureList();
+    for(Future<RebalanceResults> fr : frlist) {
+      try {
+        long waitTime = endTime - System.nanoTime();
+        RebalanceResults rr =  fr.get(waitTime, TimeUnit.NANOSECONDS);                
+        results.addDetails((RebalanceResultsImpl)rr);
+      } catch (ExecutionException e) {
+        if(e.getCause() instanceof GemFireException) {
+          throw (GemFireException) e.getCause();
+        } else if(e.getCause() instanceof InternalGemFireError) {
+          throw (InternalGemFireError) e.getCause();
+        } else {
+          throw new InternalGemFireError(e.getCause());
+        }
       }
     }
+    return results;
   }
 
   public boolean isCancelled() {
     return this.cancelled.get();
   }
 
+  private boolean isAllDone() {
+    for(Future<RebalanceResults> fr : getFutureList()) {
+      if(!fr.isDone())
+        return false;
+    }
+    return true;
+  }
+  
   public boolean isDone() {
-    return this.cancelled.get() || getFuture().isDone();
+    return this.cancelled.get() || isAllDone();
   }
   
   /**

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/f7242d23/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/control/RebalanceResultsImpl.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/control/RebalanceResultsImpl.java
b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/control/RebalanceResultsImpl.java
index a74017b..0506e4c 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/control/RebalanceResultsImpl.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/control/RebalanceResultsImpl.java
@@ -39,6 +39,20 @@ public class RebalanceResultsImpl implements RebalanceResults, Serializable
{
     totalTime += details.getTime();
   }
 
+  public void addDetails(RebalanceResultsImpl details) {
+    this.detailSet.addAll(details.detailSet);
+    totalBucketCreateBytes += details.totalBucketCreateBytes;
+    totalBucketCreateTime += details.totalBucketCreateTime;
+    totalBucketCreatesCompleted += details.totalBucketCreatesCompleted;
+    totalBucketTransferBytes += details.totalBucketTransferBytes;
+    totalBucketTransferTime += details.totalBucketTransferTime;
+    totalBucketTransfersCompleted += details.totalBucketTransfersCompleted;
+    totalPrimaryTransferTime += details.totalPrimaryTransferTime;
+    totalPrimaryTransfersCompleted += details.totalPrimaryTransfersCompleted;
+    if(details.totalTime > totalTime)
+    totalTime = details.totalTime;
+  }
+
   public Set<PartitionRebalanceInfo> getPartitionRebalanceDetails() {
     return detailSet;
   }
@@ -78,4 +92,4 @@ public class RebalanceResultsImpl implements RebalanceResults, Serializable
{
   public long getTotalTime() {
     return this.totalTime;
   }
-}
\ No newline at end of file
+}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/f7242d23/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/control/RebalanceOperationDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/control/RebalanceOperationDUnitTest.java
b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/control/RebalanceOperationDUnitTest.java
index a80cb9b..041a217 100644
--- a/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/control/RebalanceOperationDUnitTest.java
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/cache/control/RebalanceOperationDUnitTest.java
@@ -14,8 +14,11 @@ import java.util.List;
 import java.util.Properties;
 import java.util.Set;
 import java.util.TreeSet;
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.Callable;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
@@ -52,6 +55,7 @@ import com.gemstone.gemfire.internal.cache.DiskStoreImpl;
 import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
 import com.gemstone.gemfire.internal.cache.PartitionedRegion;
 import com.gemstone.gemfire.internal.cache.PartitionedRegionDataStore;
+import com.gemstone.gemfire.internal.cache.control.InternalResourceManager;
 import com.gemstone.gemfire.internal.cache.control.InternalResourceManager.ResourceObserverAdapter;
 
 import dunit.AsyncInvocation;
@@ -67,7 +71,7 @@ import dunit.VM;
 @SuppressWarnings("synthetic-access")
 public class RebalanceOperationDUnitTest extends CacheTestCase {
 
-  private static final long MAX_WAIT = 6000;
+  private static final long MAX_WAIT = 60;
   
   
   
@@ -77,9 +81,11 @@ public class RebalanceOperationDUnitTest extends CacheTestCase {
     invokeInEveryVM(new SerializableRunnable() {
       public void run() {
         InternalResourceManager.setResourceObserver(null);
+        System.clearProperty("gemfire.resource.manager.threads");
       }
     });
     InternalResourceManager.setResourceObserver(null);
+    System.clearProperty("gemfire.resource.manager.threads");
   }
 
   /**
@@ -531,27 +537,232 @@ public class RebalanceOperationDUnitTest extends CacheTestCase {
     
     
     if(!simulate) {
-    checkBucketCount(vm0, 3);
-    checkBucketCount(vm1, 3);
-    checkBucketCount(vm2, 6);
+    checkBucketCount(vm0, "region1", 3);
+    checkBucketCount(vm1, "region1", 3);
+    checkBucketCount(vm2, "region1", 6);
     }
     
     } finally {
       disconnectFromDS();
       invokeInEveryVM(new SerializableRunnable() {
         public void run() {
+          //clear the redundancy zone setting
           disconnectFromDS(); 
         }
       });
     }
   }
 
-  private void checkBucketCount(VM vm0, final int numLocalBuckets) {
+  private void createPR(String regionName){
+    Cache cache = getCache();
+    AttributesFactory attr = new AttributesFactory();
+    PartitionAttributesFactory paf = new PartitionAttributesFactory();
+    paf.setRedundantCopies(1);
+    paf.setRecoveryDelay(-1);
+    paf.setStartupRecoveryDelay(-1);
+    PartitionAttributes prAttr = paf.create();
+    attr.setPartitionAttributes(prAttr);
+    cache.createRegion(regionName, attr.create());
+  }
+  
+  private void doPuts(String regionName) {
+    Cache cache = getCache();
+    Region region = cache.getRegion(regionName);
+    region.put(Integer.valueOf(1), "A");
+    region.put(Integer.valueOf(2), "A");
+    region.put(Integer.valueOf(3), "A");
+    region.put(Integer.valueOf(4), "A");
+    region.put(Integer.valueOf(5), "A");
+    region.put(Integer.valueOf(6), "A");
+  }
+  
+  public static class ParallelRecoveryObserver extends InternalResourceManager.ResourceObserverAdapter
{
+    
+    HashSet<String> regions = new HashSet<String>();
+    private volatile boolean observerCalled;
+    private CyclicBarrier barrier;
+    
+    public ParallelRecoveryObserver(int numRegions) {
+      this.barrier = new CyclicBarrier(numRegions);
+    }
+    
+    public void observeRegion(String region) {
+      regions.add(region);
+    }
+    
+    private void checkAllRegionRecoveryOrRebalanceStarted(String rn) {
+      if(regions.contains(rn)) {
+        try {
+          barrier.await(MAX_WAIT, TimeUnit.SECONDS);
+        } catch (Exception e) {
+          fail("failed waiting for barrier", e);
+        }
+        observerCalled = true;
+      } else {
+        throw new RuntimeException("region not registered " + rn );
+      }
+    }
+    
+    public boolean isObserverCalled(){
+      return observerCalled;
+    }
+    
+    @Override
+    public void rebalancingStarted(Region region) {
+      
+      // TODO Auto-generated method stub
+      super.rebalancingStarted(region);
+      checkAllRegionRecoveryOrRebalanceStarted(region.getName());
+    }
+    
+    @Override
+    public void recoveryStarted(Region region) {
+      // TODO Auto-generated method stub
+      super.recoveryStarted(region);
+      checkAllRegionRecoveryOrRebalanceStarted(region.getName());
+    }
+  }
+  
+  public void testEnforceZoneWithMultipleRegions() {
+    Host host = Host.getHost(0);
+    VM vm0 = host.getVM(0);
+    VM vm1 = host.getVM(1);
+    VM vm2 = host.getVM(2);
+    
+    try {
+    setRedundancyZone(vm0, "A");
+    setRedundancyZone(vm1, "A");
+    
+    final DistributedMember zoneBMember = setRedundancyZone(vm2, "B");
+    
+    SerializableRunnable setRebalanceObserver = new SerializableRunnable("RebalanceObserver")
{
+      @Override
+      public void run() {
+        InternalResourceManager.setResourceObserver(new ParallelRecoveryObserver(2));   
    
+      }
+    };
+
+    SerializableRunnable createPrRegion = new SerializableRunnable("createRegion") {
+      public void run()
+      {
+        ParallelRecoveryObserver ob = (ParallelRecoveryObserver)InternalResourceManager.getResourceObserver();
+        ob.observeRegion("region1");
+        ob.observeRegion("region2");
+        createPR("region1");        
+        createPR("region2");
+                
+      }
+    };
+    
+    vm0.invoke(setRebalanceObserver);
+    //Create the region in only 1 VM
+    vm0.invoke(createPrRegion);
+    
+    //Create some buckets
+    vm0.invoke(new SerializableRunnable("createSomeBuckets") {
+      
+      public void run() {
+        doPuts("region1");
+        doPuts("region2");
+      }
+    });
+    
+    SerializableRunnable checkLowRedundancy = new SerializableRunnable("checkLowRedundancy")
{
+
+      public void run() {
+        Cache cache = getCache();
+        Region region = cache.getRegion("region1");
+        PartitionRegionInfo details = PartitionRegionHelper.getPartitionRegionInfo(region);
+        assertEquals(6, details.getCreatedBucketCount());
+        assertEquals(0,details.getActualRedundantCopies());
+        assertEquals(6,details.getLowRedundancyBucketCount());
+        
+        region = cache.getRegion("region2");
+        details = PartitionRegionHelper.getPartitionRegionInfo(region);
+        assertEquals(6, details.getCreatedBucketCount());
+        assertEquals(0,details.getActualRedundantCopies());
+        assertEquals(6,details.getLowRedundancyBucketCount());
+      }
+    };
+    
+    //make sure we can tell that the buckets have low redundancy
+    vm0.invoke(checkLowRedundancy);
+
+    //Create the region in the other VMs (should have no effect)
+    vm1.invoke(setRebalanceObserver);
+    vm1.invoke(createPrRegion);
+    vm2.invoke(setRebalanceObserver);
+    vm2.invoke(createPrRegion);
+    
+    //Make sure we still have low redundancy
+    vm0.invoke(checkLowRedundancy);
+    
+    //Now do a rebalance
+    vm0.invoke(new SerializableRunnable("simulateRebalance") {
+
+      public void run() {
+        Cache cache = getCache();
+        ResourceManager manager = cache.getResourceManager();
+        RebalanceResults results = doRebalance(false, manager);
+        //We expect to satisfy redundancy with the zone B member
+        assertEquals(12, results.getTotalBucketCreatesCompleted());
+        //2 primaries will go to vm2, leaving vm0 and vm1 with 2 primaries each
+        assertEquals(4, results.getTotalPrimaryTransfersCompleted());
+        //We actually *will* transfer 3 buckets to the other member in zone A, because that
improves
+        //the balance
+        assertEquals(6, results.getTotalBucketTransfersCompleted());
+        Set<PartitionRebalanceInfo> detailSet = results.getPartitionRebalanceDetails();
+        assertEquals(2, detailSet.size());
+        for(PartitionRebalanceInfo details : detailSet) {
+          assertEquals(6, details.getBucketCreatesCompleted());
+          assertEquals(2, details.getPrimaryTransfersCompleted());
+          assertEquals(3, details.getBucketTransfersCompleted());
+          Set<PartitionMemberInfo> afterDetails = details.getPartitionMemberDetailsAfter();
+          for(PartitionMemberInfo info : afterDetails) {
+            if(info.getDistributedMember().equals(zoneBMember)) {
+              assertEquals(6, info.getBucketCount());
+            } else {
+              assertEquals(3, info.getBucketCount());
+            }
+            assertEquals(2, info.getPrimaryCount());
+          }
+        }
+        //        assertEquals(0, details.getBucketTransferBytes());
+        verifyStats(manager, results);
+      }
+    });
+    
+    vm0.invoke(new SerializableRunnable() {
+      
+      @Override
+      public void run() {
+        assertTrue(((ParallelRecoveryObserver)InternalResourceManager.getResourceObserver()).isObserverCalled());
+      }
+    });
+    
+    checkBucketCount(vm0, "region1", 3);
+    checkBucketCount(vm1, "region1", 3);
+    checkBucketCount(vm2, "region1", 6);
+    
+    checkBucketCount(vm0, "region2", 3);
+    checkBucketCount(vm1, "region2", 3);
+    checkBucketCount(vm2, "region2", 6);
+    } finally {
+      invokeInEveryVM(new SerializableRunnable() {
+        public void run() {
+          //clear the redundancy zone setting
+          disconnectFromDS(); 
+        }
+      });
+    }
+  }
+  
+  private void checkBucketCount(VM vm0, final String regionName, final int numLocalBuckets)
{
     vm0.invoke(new SerializableRunnable("checkLowRedundancy") {
 
       public void run() {
         Cache cache = getCache();
-        PartitionedRegion region = (PartitionedRegion) cache.getRegion("region1");
+        PartitionedRegion region = (PartitionedRegion) cache.getRegion(regionName);
         assertEquals(numLocalBuckets, region.getLocalBucketsListTestOnly().size());
       }
     });
@@ -561,6 +772,7 @@ public class RebalanceOperationDUnitTest extends CacheTestCase {
   private DistributedMember setRedundancyZone(VM vm, final String zone) {
     return (DistributedMember) vm.invoke(new SerializableCallable("set redundancy zone")
{
       public Object call() {
+        System.setProperty("gemfire.resource.manager.threads", "2");
         Properties props = new Properties();
         props.setProperty(DistributionConfig.REDUNDANCY_ZONE_NAME, zone);
         DistributedSystem system = getSystem(props);
@@ -2541,7 +2753,7 @@ public class RebalanceOperationDUnitTest extends CacheTestCase {
     assertEquals(0, stats.getRebalanceBucketCreatesInProgress());
     assertEquals(results.getTotalBucketCreatesCompleted(), stats.getRebalanceBucketCreatesCompleted());
     assertEquals(0, stats.getRebalanceBucketCreatesFailed());
-    assertEquals(results.getTotalBucketCreateTime(), TimeUnit.NANOSECONDS.toMillis(stats.getRebalanceBucketCreateTime()));
+//    assertEquals(results.getTotalBucketCreateTime(), TimeUnit.NANOSECONDS.toMillis(stats.getRebalanceBucketCreateTime()));
     assertEquals(results.getTotalBucketCreateBytes(), stats.getRebalanceBucketCreateBytes());
     assertEquals(0, stats.getRebalanceBucketTransfersInProgress());
     assertEquals(results.getTotalBucketTransfersCompleted(), stats.getRebalanceBucketTransfersCompleted());
@@ -2551,7 +2763,7 @@ public class RebalanceOperationDUnitTest extends CacheTestCase {
     assertEquals(0, stats.getRebalancePrimaryTransfersInProgress());
     assertEquals(results.getTotalPrimaryTransfersCompleted(), stats.getRebalancePrimaryTransfersCompleted());
     assertEquals(0, stats.getRebalancePrimaryTransfersFailed());
-    assertEquals(results.getTotalPrimaryTransferTime(), TimeUnit.NANOSECONDS.toMillis(stats.getRebalancePrimaryTransferTime()));
+//    assertEquals(results.getTotalPrimaryTransferTime(), TimeUnit.NANOSECONDS.toMillis(stats.getRebalancePrimaryTransferTime()));
   }
   
   private Set<Integer> getBucketList(final String regionName, VM vm0) {


Mime
View raw message