<NME> RebalanceUtilsTest.java
<BEF> /*
 * Copyright 2011-2013 LinkedIn, Inc
 * 
 * Licensed 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 voldemort.utils;

import java.util.Arrays;

import junit.framework.TestCase;
import voldemort.client.rebalance.RebalanceStealInfo;

public class RebalanceUtilsTest extends TestCase {

    public void testRebalanceStealInfo() {
        RebalanceStealInfo info = new RebalanceStealInfo(0,
                                                         1,
        if(verify)
            assertEquals(interimC, derivedInterim2);

        RebalanceUtils.validateCurrentFinalCluster(currentC, finalC);
        RebalanceUtils.validateCurrentInterimCluster(currentC, interimC);
        RebalanceUtils.validateInterimFinalCluster(interimC, finalC);
    }

    private void doClusterTransformation(Cluster currentC, Cluster interimC, Cluster finalC) {
        doClusterTransformationBase(currentC, interimC, finalC, false);
    }

    public void doClusterTransformationAndVerification(Cluster currentC,
                                                       Cluster interimC,
                                                       Cluster finalC) {
        doClusterTransformationBase(currentC, interimC, finalC, true);
    }

    @Test
    public void testClusterTransformationAndVerification() {
        // Two-zone cluster: no-op
        doClusterTransformationAndVerification(ClusterTestUtils.getZZCluster(),
                                               ClusterTestUtils.getZZCluster(),
                                               ClusterTestUtils.getZZCluster());

        // Two-zone cluster: rebalance
        doClusterTransformationAndVerification(ClusterTestUtils.getZZCluster(),
                                               ClusterTestUtils.getZZCluster(),
                                               ClusterTestUtils.getZZClusterWithSwappedPartitions());

        // Two-zone cluster: cluster expansion
        doClusterTransformationAndVerification(ClusterTestUtils.getZZCluster(),
                                               ClusterTestUtils.getZZClusterWithNN(),
                                               ClusterTestUtils.getZZClusterWithPP());

        // Three-zone cluster: no-op
        doClusterTransformationAndVerification(ClusterTestUtils.getZZZCluster(),
                                               ClusterTestUtils.getZZZCluster(),
                                               ClusterTestUtils.getZZZCluster());

        // Three-zone cluster: rebalance
        doClusterTransformationAndVerification(ClusterTestUtils.getZZZCluster(),
                                               ClusterTestUtils.getZZZCluster(),
                                               ClusterTestUtils.getZZZClusterWithSwappedPartitions());

        // Three-zone cluster: cluster expansion
        doClusterTransformationAndVerification(ClusterTestUtils.getZZZCluster(),
                                               ClusterTestUtils.getZZZClusterWithNNN(),
                                               ClusterTestUtils.getZZZClusterWithPPP());

        doClusterTransformationAndVerification(ClusterTestUtils.getZZCluster(),
                                               ClusterTestUtils.getZZECluster(),
                                               ClusterTestUtils.getZZEClusterXXP());
    }

    @Test
    public void testClusterTransformationAndVerificationExceptions() {
        boolean excepted;

        // Two-zone cluster: rebalance with extra partitions in interim cluster
        excepted = false;
        try {
            doClusterTransformation(ClusterTestUtils.getZZCluster(),
                                    ClusterTestUtils.getZZClusterWithExtraPartitions(),
                                    ClusterTestUtils.getZZClusterWithSwappedPartitions());
        } catch(VoldemortException ve) {
            excepted = true;
        }
        assertTrue(excepted);

        // Two-zone cluster: rebalance with extra partitions in final
        excepted = false;
        try {
            doClusterTransformation(ClusterTestUtils.getZZCluster(),
                                    ClusterTestUtils.getZZCluster(),
                                    ClusterTestUtils.getZZClusterWithExtraPartitions());
        } catch(VoldemortException ve) {
            excepted = true;
        }
        assertTrue(excepted);

        // Two-zone cluster: node ids swapped in interim cluster
        excepted = false;
        try {
            doClusterTransformation(ClusterTestUtils.getZZCluster(),
                                    ClusterTestUtils.getZZClusterWithNNWithSwappedNodeIds(),
                                    ClusterTestUtils.getZZClusterWithPP());
        } catch(VoldemortException ve) {
            excepted = true;
        }
        assertTrue(excepted);

        // Two-zone cluster: node ids swapped in final is OK because this is the
        // same as partitions being migrated among nodes.
        excepted = false;
        try {
            doClusterTransformation(ClusterTestUtils.getZZCluster(),
                                    ClusterTestUtils.getZZClusterWithNN(),
                                    ClusterTestUtils.getZZClusterWithPPWithSwappedNodeIds());
        } catch(VoldemortException ve) {
            excepted = true;
        }
        assertFalse(excepted);

        // Two-zone cluster: too many node ids in final
        excepted = false;
        try {
            doClusterTransformation(ClusterTestUtils.getZZCluster(),
                                    ClusterTestUtils.getZZClusterWithNN(),
                                    ClusterTestUtils.getZZClusterWithPPWithTooManyNodes());
        } catch(VoldemortException ve) {
            excepted = true;
        }
        assertTrue(excepted);
    }
}

<MSG> Rebalance unit test start passing.

<DFF> @@ -16,13 +16,84 @@
 
 package voldemort.utils;
 
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileReader;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.List;
+import java.util.Queue;
 
 import junit.framework.TestCase;
+import voldemort.ServerTestUtils;
 import voldemort.client.rebalance.RebalanceStealInfo;
+import voldemort.cluster.Cluster;
+import voldemort.cluster.Node;
+import voldemort.store.StoreDefinition;
+import voldemort.xml.StoreDefinitionsMapper;
 
 public class RebalanceUtilsTest extends TestCase {
 
+    private static String testStoreName = "test-replication-memory";
+    private static String storeDefFile = "test/common/voldemort/config/stores.xml";
+
+    private Cluster currentCluster;
+    private Cluster targetCluster;
+    List<StoreDefinition> storeDefList;
+
+    @Override
+    public void setUp() {
+        currentCluster = ServerTestUtils.getLocalCluster(2, new int[][] {
+                { 0, 1, 2, 3, 4, 5, 6, 7, 8 }, {}, {}, {} });
+
+        targetCluster = ServerTestUtils.getLocalCluster(2, new int[][] { { 0, 1, 4, 5, 6, 7, 8 },
+                { 2, 3 } });
+
+        try {
+            storeDefList = new StoreDefinitionsMapper().readStoreList(new FileReader(new File(storeDefFile)));
+        } catch(FileNotFoundException e) {
+            throw new RuntimeException("Failed to find storeDefFile:" + storeDefFile, e);
+        }
+    }
+
+    public void testRebalancePlan() {
+        Queue<Pair<Integer, List<RebalanceStealInfo>>> rebalancePlan = RebalanceUtils.getRebalanceTaskQueue(currentCluster,
+                                                                                                            targetCluster,
+                                                                                                            Arrays.asList(testStoreName));
+        int[][] stealList = { {}, { 2, 3 } };
+
+        // the rebalancing plan should have exactly 3 entries.
+        assertEquals("There should be three node rebalancing", 1, rebalancePlan.size());
+        for(Pair<Integer, List<RebalanceStealInfo>> rebalanceInfo: rebalancePlan) {
+            assertEquals("rebalanceInfo should have exactly one item", 1, rebalanceInfo.getSecond()
+                                                                                       .size());
+            RebalanceStealInfo expected = new RebalanceStealInfo(rebalanceInfo.getFirst(),
+                                                                 0,
+                                                                 listFromArray(stealList[rebalanceInfo.getFirst()]),
+                                                                 Arrays.asList(testStoreName),
+                                                                 0);
+
+            assertEquals("rebalanceStealInfo should match",
+                         expected.toJsonString(),
+                         rebalanceInfo.getSecond().get(0).toJsonString());
+        }
+    }
+
+    public void testUpdateCluster() {
+        Cluster updatedCluster = RebalanceUtils.updateCluster(currentCluster,
+                                                              new ArrayList<Node>(targetCluster.getNodes()));
+        assertEquals("updated cluster should match targetCluster", updatedCluster, targetCluster);
+    }
+
+    private List<Integer> listFromArray(int[] array) {
+        List<Integer> list = new ArrayList<Integer>();
+        for(int x: array) {
+            list.add(x);
+        }
+
+        return list;
+    }
+
     public void testRebalanceStealInfo() {
         RebalanceStealInfo info = new RebalanceStealInfo(0,
                                                          1,

{'commit': 'diff --git a/src/java/log4j.properties b/src/java/log4j.properties\nindex e568e44f0..461ba65c9 100755\n--- a/src/java/log4j.properties\n+++ b/src/java/log4j.properties\n@@ -13,7 +13,8 @@ log4j.logger.httpclient.wire=INFO\n log4j.logger.org.mortbay.log=WARN\n log4j.logger.voldemort.store.routed=INFO\n log4j.logger.voldemort.server.niosocket=INFO\n-log4j.logger.voldemort.utils=INFO\n+log4j.logger.voldemort.utils=DEBUG\n log4j.logger.voldemort.client.rebalance=INFO\n-log4j.logger.voldemort.server.protocol.admin=INFO\n+log4j.logger.voldemort.server.protocol.admin=DEBUG\n+log4j.logger.voldemort.client.rebalance=DEBUG\n log4j.logger.voldemort.server=INFO\ndiff --git a/src/java/voldemort/client/rebalance/RebalanceClient.java b/src/java/voldemort/client/rebalance/RebalanceClient.java\nindex 957034f9b..8f3576e72 100644\n--- a/src/java/voldemort/client/rebalance/RebalanceClient.java\n+++ b/src/java/voldemort/client/rebalance/RebalanceClient.java\n@@ -54,13 +54,12 @@ public class RebalanceClient {\n      * via {@link RedirectingStore}<br>\n      * \n      * \n-     * @param currentCluster: currentCluster configuration.\n      * @param targetCluster: target Cluster configuration\n+     * @param storeList: All stores which should be rebalanced.\n      */\n-    public void rebalance(final Cluster currentCluster,\n-                          final Cluster targetCluster,\n-                          final List<String> storeList) {\n-        // update adminClient with currentCluster\n+    public void rebalance(final Cluster targetCluster, final List<String> storeList) {\n+        Cluster currentCluster = RebalanceUtils.getLatestCluster(new ArrayList<Integer>(),\n+                                                                 adminClient).getValue();\n         adminClient.setAdminClientCluster(currentCluster);\n \n         if(!RebalanceUtils.getClusterRebalancingToken()) {\n@@ -79,6 +78,8 @@ public class RebalanceClient {\n                 public void run() {\n                     // pick one node to rebalance from queue\n                     while(!rebalanceTaskQueue.isEmpty()) {\n+                        logger.debug("rebalanceTaskQueue size:" + rebalanceTaskQueue.size());\n+\n                         Pair<Integer, List<RebalanceStealInfo>> rebalanceTask = rebalanceTaskQueue.poll();\n                         if(null != rebalanceTask) {\n                             int stealerNodeId = rebalanceTask.getFirst();\n@@ -109,6 +110,7 @@ public class RebalanceClient {\n                             }\n                         }\n                     }\n+                    logger.debug("Thread run() finished:\\n");\n                 }\n             });\n         }// for (nThreads ..\n@@ -133,6 +135,7 @@ public class RebalanceClient {\n     private void executorShutDown(ExecutorService executorService) {\n         try {\n             executorService.shutdown();\n+            executorService.awaitTermination(24 * 60 * 60, TimeUnit.SECONDS);\n         } catch(Exception e) {\n             logger.warn("Error while stoping executor service .. ", e);\n         }\n@@ -160,8 +163,10 @@ public class RebalanceClient {\n      * @param stealPartitionsMap\n      * @param stealerNodeId\n      * @param rebalanceStealInfo\n+     * @throws Exception\n      */\n-    void commitClusterChanges(Node stealerNode, RebalanceStealInfo rebalanceStealInfo) {\n+    void commitClusterChanges(Node stealerNode, RebalanceStealInfo rebalanceStealInfo)\n+            throws Exception {\n         synchronized(adminClient) {\n             Versioned<Cluster> latestVersionedCluster = RebalanceUtils.getLatestCluster(Arrays.asList(stealerNode.getId(),\n                                                                                                       rebalanceStealInfo.getDonorId()),\n@@ -186,8 +191,6 @@ public class RebalanceClient {\n                 // set new cluster in adminClient\n                 adminClient.setAdminClientCluster(newCluster);\n             } catch(Exception e) {\n-                logger.error("Failed to commit rebalance on node:" + stealerNode.getId()\n-                             + " REVERTING cluster changes ...", e);\n                 // revert cluster changes.\n                 latestClock.incrementVersion(stealerNode.getId(), System.currentTimeMillis());\n                 RebalanceUtils.propagateCluster(adminClient,\n@@ -195,7 +198,7 @@ public class RebalanceClient {\n                                                 latestClock,\n                                                 new ArrayList<Integer>());\n \n-                throw new VoldemortException(e);\n+                throw e;\n             }\n         }\n     }\n@@ -212,7 +215,7 @@ public class RebalanceClient {\n     void attemptRebalanceSubTask(RebalanceStealInfo rebalanceSubTask) {\n         boolean success = true;\n         List<String> rebalanceStoreList = ImmutableList.copyOf(rebalanceSubTask.getUnbalancedStoreList());\n-\n+        List<String> unbalancedStoreList = new ArrayList<String>(rebalanceStoreList);\n         for(String storeName: rebalanceStoreList) {\n             try {\n                 int rebalanceAsyncId = adminClient.rebalanceNode(storeName, rebalanceSubTask);\n@@ -222,7 +225,8 @@ public class RebalanceClient {\n                                               24 * 60 * 60,\n                                               TimeUnit.SECONDS);\n                 // remove store from rebalance list\n-                rebalanceSubTask.getUnbalancedStoreList().remove(storeName);\n+                unbalancedStoreList.remove(storeName);\n+                rebalanceSubTask.setUnbalancedStoreList(unbalancedStoreList);\n             } catch(Exception e) {\n                 logger.warn("rebalanceSubTask:" + rebalanceSubTask + " failed for store:"\n                             + storeName, e);\ndiff --git a/src/java/voldemort/client/rebalance/RebalanceCommandShell.java b/src/java/voldemort/client/rebalance/RebalanceCommandShell.java\nindex f78dfb575..f09a38388 100644\n--- a/src/java/voldemort/client/rebalance/RebalanceCommandShell.java\n+++ b/src/java/voldemort/client/rebalance/RebalanceCommandShell.java\n@@ -36,6 +36,6 @@ public class RebalanceCommandShell {\n         for(StoreDefinition storeDef: storesList)\n             storeNames.add(storeDef.getName());\n \n-        rebalanceClient.rebalance(currentCluster, targetCluster, storeNames);\n+        rebalanceClient.rebalance(targetCluster, storeNames);\n     }\n }\n\\ No newline at end of file\ndiff --git a/src/java/voldemort/client/rebalance/RebalanceStealInfo.java b/src/java/voldemort/client/rebalance/RebalanceStealInfo.java\nindex 50b7e5d77..6549871ec 100644\n--- a/src/java/voldemort/client/rebalance/RebalanceStealInfo.java\n+++ b/src/java/voldemort/client/rebalance/RebalanceStealInfo.java\n@@ -16,7 +16,7 @@ public class RebalanceStealInfo {\n     private final int stealerId;\n     private final int donorId;\n     private final List<Integer> partitionList;\n-    private final List<String> unbalancedStoreList;\n+    private List<String> unbalancedStoreList;\n \n     private int attempt;\n \n@@ -33,6 +33,7 @@ public class RebalanceStealInfo {\n         this.unbalancedStoreList = unbalancedStorList;\n     }\n \n+    @SuppressWarnings("unchecked")\n     public RebalanceStealInfo(String line) {\n         try {\n             JsonReader reader = new JsonReader(new StringReader(line));\n@@ -72,6 +73,10 @@ public class RebalanceStealInfo {\n         return unbalancedStoreList;\n     }\n \n+    public void setUnbalancedStoreList(List<String> storeList) {\n+        this.unbalancedStoreList = storeList;\n+    }\n+\n     @Override\n     public String toString() {\n         return "RebalancingStealInfo(" + getStealerId() + " <--- " + getDonorId() + " partitions:"\ndiff --git a/src/java/voldemort/routing/ConsistentRoutingStrategy.java b/src/java/voldemort/routing/ConsistentRoutingStrategy.java\nindex e8dc86226..3889dd4ed 100644\n--- a/src/java/voldemort/routing/ConsistentRoutingStrategy.java\n+++ b/src/java/voldemort/routing/ConsistentRoutingStrategy.java\n@@ -62,7 +62,7 @@ public class ConsistentRoutingStrategy implements RoutingStrategy {\n             for(Integer partition: n.getPartitionIds()) {\n                 if(m.containsKey(partition))\n                     throw new IllegalArgumentException("Duplicate partition id " + partition\n-                                                       + " in cluster configuration.");\n+                                                       + " in cluster configuration " + nodes);\n                 m.put(partition, n);\n             }\n         }\ndiff --git a/src/java/voldemort/server/protocol/admin/ProtoBuffAdminServiceRequestHandler.java b/src/java/voldemort/server/protocol/admin/ProtoBuffAdminServiceRequestHandler.java\nindex 8dc802ecb..62f20046d 100644\n--- a/src/java/voldemort/server/protocol/admin/ProtoBuffAdminServiceRequestHandler.java\n+++ b/src/java/voldemort/server/protocol/admin/ProtoBuffAdminServiceRequestHandler.java\n@@ -96,7 +96,6 @@ public class ProtoBuffAdminServiceRequestHandler implements RequestHandler {\n         ByteUtils.read(inputStream, input);\n         request.mergeFrom(input);\n \n-        logger.debug("AdminClient recieved request:" + request);\n         switch(request.getType()) {\n             case GET_METADATA:\n                 ProtoUtils.writeMessage(outputStream, handleGetMetadata(request.getGetMetadata()));\ndiff --git a/src/java/voldemort/utils/RebalanceUtils.java b/src/java/voldemort/utils/RebalanceUtils.java\nindex 232e974de..bcc27308b 100644\n--- a/src/java/voldemort/utils/RebalanceUtils.java\n+++ b/src/java/voldemort/utils/RebalanceUtils.java\n@@ -163,7 +163,7 @@ public class RebalanceUtils {\n         List<Integer> donorPartitionList = new ArrayList<Integer>(donorNode.getPartitionIds());\n \n         for(int p: partitionList) {\n-            donorPartitionList.remove(p);\n+            removePartition(donorPartitionList, p);\n             if(!stealerPartitionList.contains(p))\n                 stealerPartitionList.add(p);\n         }\n@@ -176,7 +176,17 @@ public class RebalanceUtils {\n         stealerNode = updateNode(stealerNode, stealerPartitionList);\n         donorNode = updateNode(donorNode, donorPartitionList);\n \n-        return updateCluster(cluster, Arrays.asList(stealerNode, donorNode));\n+        Cluster updatedCluster = updateCluster(cluster, Arrays.asList(stealerNode, donorNode));\n+        logger.debug("currentCluster: " + cluster + " updatedCluster:" + updatedCluster);\n+        return updatedCluster;\n+    }\n+\n+    private static void removePartition(List<Integer> donorPartitionList, int partition) {\n+        for(int i = 0; i < donorPartitionList.size(); i++) {\n+            if(partition == donorPartitionList.get(i)) {\n+                donorPartitionList.remove(i);\n+            }\n+        }\n     }\n \n     public static Cluster updateCluster(Cluster currentCluster, List<Node> updatedNodeList) {\n@@ -273,8 +283,8 @@ public class RebalanceUtils {\n                 logger.debug("Updating remote node:" + nodeId + " with cluster:" + cluster);\n                 adminClient.updateRemoteCluster(node.getId(), cluster, clock);\n             } catch(Exception e) {\n-                throw new VoldemortException("Failed to copy new cluster.xml(" + cluster\n-                                             + ") on required node:" + node, e);\n+                throw new VoldemortException("Failed to copy updated cluster.xml:" + cluster\n+                                             + " on required node:" + node, e);\n             }\n         }\n \ndiff --git a/test/common/voldemort/ServerTestUtils.java b/test/common/voldemort/ServerTestUtils.java\nindex 705ae5729..d34abda69 100644\n--- a/test/common/voldemort/ServerTestUtils.java\n+++ b/test/common/voldemort/ServerTestUtils.java\n@@ -53,6 +53,7 @@ import voldemort.server.socket.SocketService;\n import voldemort.store.Store;\n import voldemort.store.StoreDefinition;\n import voldemort.store.StoreDefinitionBuilder;\n+import voldemort.store.UnreachableStoreException;\n import voldemort.store.http.HttpStore;\n import voldemort.store.memory.InMemoryStorageConfiguration;\n import voldemort.store.memory.InMemoryStorageEngine;\n@@ -155,9 +156,16 @@ public class ServerTestUtils {\n     }\n \n     public static SocketStore getSocketStore(String storeName, int port, RequestFormatType type) {\n+        return getSocketStore(storeName, "localhost", port, RequestFormatType.VOLDEMORT_V1);\n+    }\n+\n+    public static SocketStore getSocketStore(String storeName,\n+                                             String host,\n+                                             int port,\n+                                             RequestFormatType type) {\n         SocketPool socketPool = new SocketPool(2, 10000, 100000, 32 * 1024);\n         return new SocketStore(storeName,\n-                               new SocketDestination("localhost", port, type),\n+                               new SocketDestination(host, port, type),\n                                socketPool,\n                                false);\n     }\n@@ -398,7 +406,37 @@ public class ServerTestUtils {\n     public static VoldemortServer startVoldemortServer(VoldemortConfig config, Cluster cluster) {\n         VoldemortServer server = new VoldemortServer(config, cluster);\n         server.start();\n+\n+        ServerTestUtils.waitForServerStart(server.getIdentityNode());\n+        // wait till server start or throw exception\n         return server;\n     }\n \n+    public static void waitForServerStart(Node node) {\n+        boolean success = false;\n+        int retries = 10;\n+        Store<ByteArray, ?> store = null;\n+        while(retries-- > 0) {\n+            store = ServerTestUtils.getSocketStore(MetadataStore.METADATA_STORE_NAME,\n+                                                   node.getSocketPort());\n+            try {\n+                store.get(new ByteArray(MetadataStore.CLUSTER_KEY.getBytes()));\n+                success = true;\n+            } catch(UnreachableStoreException e) {\n+                store.close();\n+                store = null;\n+                System.out.println("UnreachableSocketStore sleeping will try again " + retries\n+                                   + " times.");\n+                try {\n+                    Thread.sleep(1000);\n+                } catch(InterruptedException e1) {\n+                    // ignore\n+                }\n+            }\n+        }\n+\n+        store.close();\n+        if(!success)\n+            throw new RuntimeException("Failed to connect with server:" + node);\n+    }\n }\ndiff --git a/test/common/voldemort/utils/ServerJVMTestUtils.java b/test/common/voldemort/utils/ServerJVMTestUtils.java\nindex 1839ffd4d..27d516128 100644\n--- a/test/common/voldemort/utils/ServerJVMTestUtils.java\n+++ b/test/common/voldemort/utils/ServerJVMTestUtils.java\n@@ -14,9 +14,6 @@ import voldemort.TestUtils;\n import voldemort.cluster.Cluster;\n import voldemort.cluster.Node;\n import voldemort.server.VoldemortConfig;\n-import voldemort.store.Store;\n-import voldemort.store.UnreachableStoreException;\n-import voldemort.store.metadata.MetadataStore;\n import voldemort.xml.ClusterMapper;\n \n /**\n@@ -33,7 +30,7 @@ public class ServerJVMTestUtils {\n         String command = "java  voldemort.server.VoldemortServer " + voldemortHome;\n         // System.out.println("command:" + command + " env:" + env);\n         Process process = Runtime.getRuntime().exec(command, env.toArray(new String[0]));\n-        waitForServerStart(node);\n+        ServerTestUtils.waitForServerStart(node);\n         startOutputErrorConsumption(process);\n         return process;\n     }\n@@ -71,34 +68,6 @@ public class ServerJVMTestUtils {\n         }).start();\n     }\n \n-    public static void waitForServerStart(Node node) {\n-        boolean success = false;\n-        int retries = 10;\n-        Store<ByteArray, ?> store = null;\n-        while(retries-- > 0) {\n-            store = ServerTestUtils.getSocketStore(MetadataStore.METADATA_STORE_NAME,\n-                                                   node.getSocketPort());\n-            try {\n-                store.get(new ByteArray(MetadataStore.CLUSTER_KEY.getBytes()));\n-                success = true;\n-            } catch(UnreachableStoreException e) {\n-                store.close();\n-                store = null;\n-                System.out.println("UnreachableSocketStore sleeping will try again " + retries\n-                                   + " times.");\n-                try {\n-                    Thread.sleep(1000);\n-                } catch(InterruptedException e1) {\n-                    // ignore\n-                }\n-            }\n-        }\n-\n-        store.close();\n-        if(!success)\n-            throw new RuntimeException("Failed to connect with server:" + node);\n-    }\n-\n     public static void StopServerJVM(Process server) {\n         System.out.println("killing process" + server);\n         server.destroy();\ndiff --git a/test/unit/voldemort/client/rebalance/AbstractRebalanceTest.java b/test/unit/voldemort/client/rebalance/AbstractRebalanceTest.java\nindex 5addfff2f..1eb331812 100644\n--- a/test/unit/voldemort/client/rebalance/AbstractRebalanceTest.java\n+++ b/test/unit/voldemort/client/rebalance/AbstractRebalanceTest.java\n@@ -1,5 +1,10 @@\n package voldemort.client.rebalance;\n \n+import java.io.File;\n+import java.io.FileNotFoundException;\n+import java.io.FileReader;\n+import java.io.IOException;\n+import java.util.Arrays;\n import java.util.HashMap;\n import java.util.List;\n import java.util.Map;\n@@ -7,66 +12,151 @@ import java.util.Map.Entry;\n \n import junit.framework.TestCase;\n import voldemort.ServerTestUtils;\n-import voldemort.client.protocol.admin.AdminClient;\n+import voldemort.client.protocol.RequestFormatType;\n+import voldemort.client.protocol.admin.AdminClientConfig;\n import voldemort.cluster.Cluster;\n import voldemort.cluster.Node;\n import voldemort.routing.ConsistentRoutingStrategy;\n import voldemort.routing.RoutingStrategy;\n+import voldemort.store.InvalidMetadataException;\n import voldemort.store.Store;\n+import voldemort.store.StoreDefinition;\n+import voldemort.store.socket.SocketStore;\n import voldemort.utils.ByteArray;\n import voldemort.utils.ByteUtils;\n import voldemort.versioning.VectorClock;\n import voldemort.versioning.Versioned;\n+import voldemort.xml.StoreDefinitionsMapper;\n \n public abstract class AbstractRebalanceTest extends TestCase {\n \n+    private static final int NUM_KEYS = 100000;\n     private static String testStoreName = "test-replication-memory";\n+    private static String storeDefFile = "test/common/voldemort/config/stores.xml";\n+\n     private Cluster currentCluster;\n     private Cluster targetCluster;\n+    List<StoreDefinition> storeDefList;\n+    HashMap<ByteArray, byte[]> testEntries;\n \n     @Override\n     public void setUp() {\n-        currentCluster = ServerTestUtils.getLocalCluster(4, new int[][] {\n-                { 0, 1, 2, 3, 4, 5, 6, 7, 8 }, {}, {}, {} });\n+        currentCluster = ServerTestUtils.getLocalCluster(2, new int[][] {\n+                { 0, 1, 2, 3, 4, 5, 6, 7, 8 }, {} });\n+\n+        targetCluster = ServerTestUtils.getLocalCluster(2, new int[][] { { 0, 1, 4, 5, 6, 7, 8 },\n+                { 2, 3 } });\n+\n+        testEntries = ServerTestUtils.createRandomKeyValuePairs(NUM_KEYS);\n+\n+        try {\n+            storeDefList = new StoreDefinitionsMapper().readStoreList(new FileReader(new File(storeDefFile)));\n+        } catch(FileNotFoundException e) {\n+            throw new RuntimeException("Failed to find storeDefFile:" + storeDefFile, e);\n+        }\n+    }\n+\n+    @Override\n+    public void tearDown() {\n+\n+    }\n+\n+    protected abstract Cluster startServers(Cluster cluster,\n+                                            List<StoreDefinition> storeDefs,\n+                                            List<Integer> nodeToStart) throws IOException;\n+\n+    protected abstract void stopServer(List<Integer> nodesToStop) throws IOException;\n+\n+    protected String getStoresXmlFile() {\n+        return storeDefFile;\n+    }\n \n-        targetCluster = ServerTestUtils.getLocalCluster(4, new int[][] { { 0, 1 }, { 2, 3 },\n-                { 4, 5 }, { 6, 7 } });\n+    private void populateData(Cluster cluster, List<Integer> nodeList) {\n+\n+        // Create SocketStores for each Node first\n+        Map<Integer, Store<ByteArray, byte[]>> storeMap = new HashMap<Integer, Store<ByteArray, byte[]>>();\n+        for(int nodeId: nodeList) {\n+            Node node = cluster.getNodeById(nodeId);\n+            storeMap.put(nodeId, ServerTestUtils.getSocketStore(testStoreName,\n+                                                                node.getHost(),\n+                                                                node.getSocketPort(),\n+                                                                RequestFormatType.PROTOCOL_BUFFERS));\n+        }\n+\n+        RoutingStrategy routing = new ConsistentRoutingStrategy(cluster.getNodes(), 1);\n+        for(Entry<ByteArray, byte[]> entry: testEntries.entrySet()) {\n+            int masterNode = routing.routeRequest(entry.getKey().get()).get(0).getId();\n+            if(nodeList.contains(masterNode)) {\n+                storeMap.get(masterNode).put(entry.getKey(), new Versioned(entry.getValue()));\n+            }\n+        }\n+\n+        // close all socket stores\n+        for(Store store: storeMap.values()) {\n+            store.close();\n+        }\n     }\n \n-    protected abstract AdminClient getAdminClient();\n+    private String getBootstrapUrl(Cluster cluster, int nodeId) {\n+        Node node = cluster.getNodeById(nodeId);\n+        return "tcp://" + node.getHost() + ":" + node.getSocketPort();\n+    }\n+\n+    // test a single rebalancing.\n+    public void testSingleRebalance() throws IOException {\n+        // start servers 0 , 1 only\n+        Cluster updatedCluster = startServers(currentCluster, storeDefList, Arrays.asList(0, 1));\n \n-    protected abstract Map<Integer, Node> startServers(Cluster cluster, List<Integer> nodeToStart);\n+        try {\n+            // generate data at node 0 only.\n+            populateData(updatedCluster, Arrays.asList(0));\n \n-    protected abstract void stopServer(List<Integer> nodesToStop);\n+            RebalanceClient rebalanceClient = new RebalanceClient(getBootstrapUrl(currentCluster, 0),\n+                                                                  1,\n+                                                                  new AdminClientConfig());\n \n-    public void testSingleRebalance() {}\n+            rebalanceClient.rebalance(targetCluster, Arrays.asList(testStoreName));\n \n-    private void checkGetEntries(HashMap<ByteArray, byte[]> entryMap,\n-                                 int nodeId,\n-                                 String storeName,\n+            checkGetEntries(updatedCluster.getNodeById(1),\n+                            Arrays.asList(0, 1, 4, 5, 6, 7, 8),\n+                            Arrays.asList(2, 3));\n+        } finally {\n+            // stop servers\n+            stopServer(Arrays.asList(0, 1));\n+        }\n+    }\n+\n+    private void checkGetEntries(Node node,\n                                  List<Integer> unavailablePartitions,\n                                  List<Integer> availablePartitions) {\n         RoutingStrategy routing = new ConsistentRoutingStrategy(targetCluster.getNodes(), 1);\n-        Store<ByteArray, byte[]> store = null;\n \n-        for(Entry<ByteArray, byte[]> entry: entryMap.entrySet()) {\n+        SocketStore store = ServerTestUtils.getSocketStore(testStoreName,\n+                                                           node.getHost(),\n+                                                           node.getSocketPort(),\n+                                                           RequestFormatType.PROTOCOL_BUFFERS);\n+\n+        for(Entry<ByteArray, byte[]> entry: testEntries.entrySet()) {\n             List<Integer> partitions = routing.getPartitionList(entry.getKey().get());\n-            List<Versioned<byte[]>> values = store.get(entry.getKey());\n \n             if(unavailablePartitions.containsAll(partitions)) {\n-                assertEquals("Store.get() for non-available partitions:" + partitions\n-                             + " should return empty list.", 0, values.size());\n+                try {\n+                    List<Versioned<byte[]>> value = store.get(entry.getKey());\n+                    fail();\n+                } catch(InvalidMetadataException e) {\n+                    // ignore.\n+                }\n             } else if(availablePartitions.containsAll(partitions)) {\n+                List<Versioned<byte[]>> values = store.get(entry.getKey());\n+\n                 // expecting exactly one version\n-                assertEquals("Store.get() for non-available partitions:" + partitions\n-                             + " should return non-empty list.", 1, values.size());\n+                assertEquals("Expecting exactly one version", 1, values.size());\n+                Versioned<byte[]> value = values.get(0);\n                 // check version matches (expecting base version for all)\n-                assertEquals("Value version should match", new VectorClock(), values.get(0)\n-                                                                                    .getVersion());\n+                assertEquals("Value version should match", new VectorClock(), value.getVersion());\n                 // check value matches.\n                 assertEquals("Value bytes should match", 0, ByteUtils.compare(entry.getValue(),\n-                                                                              values.get(0)\n-                                                                                    .getValue()));\n+                                                                              value.getValue()));\n             }\n         }\n     }\ndiff --git a/test/unit/voldemort/client/rebalance/RebalanceTest.java b/test/unit/voldemort/client/rebalance/RebalanceTest.java\nnew file mode 100644\nindex 000000000..a6bb00306\n--- /dev/null\n+++ b/test/unit/voldemort/client/rebalance/RebalanceTest.java\n@@ -0,0 +1,48 @@\n+package voldemort.client.rebalance;\n+\n+import java.io.IOException;\n+import java.util.HashMap;\n+import java.util.List;\n+import java.util.Map;\n+\n+import voldemort.ServerTestUtils;\n+import voldemort.TestUtils;\n+import voldemort.cluster.Cluster;\n+import voldemort.server.VoldemortServer;\n+import voldemort.store.StoreDefinition;\n+\n+/**\n+ * Start VoldemortServer locally using ServerTestUtils and run rebalancing\n+ * tests.\n+ * \n+ * @author bbansal\n+ * \n+ */\n+public class RebalanceTest extends AbstractRebalanceTest {\n+\n+    Map<Integer, VoldemortServer> serverMap = new HashMap<Integer, VoldemortServer>();\n+\n+    @Override\n+    protected Cluster startServers(Cluster cluster,\n+                                   List<StoreDefinition> storeDefs,\n+                                   List<Integer> nodeToStart) throws IOException {\n+        for(int node: nodeToStart) {\n+            VoldemortServer server = ServerTestUtils.startVoldemortServer(ServerTestUtils.createServerConfig(node,\n+                                                                                                             TestUtils.createTempDir()\n+                                                                                                                      .getAbsolutePath(),\n+                                                                                                             null,\n+                                                                                                             getStoresXmlFile()),\n+                                                                          cluster);\n+            serverMap.put(node, server);\n+        }\n+\n+        return cluster;\n+    }\n+\n+    @Override\n+    protected void stopServer(List<Integer> nodesToStop) throws IOException {\n+        for(int node: nodesToStop) {\n+            ServerTestUtils.stopVoldemortServer(serverMap.get(node));\n+        }\n+    }\n+}\n\\ No newline at end of file\ndiff --git a/test/unit/voldemort/utils/RebalanceUtilsTest.java b/test/unit/voldemort/utils/RebalanceUtilsTest.java\nindex 98e86cada..ac8f8d33b 100644\n--- a/test/unit/voldemort/utils/RebalanceUtilsTest.java\n+++ b/test/unit/voldemort/utils/RebalanceUtilsTest.java\n@@ -16,13 +16,84 @@\n \n package voldemort.utils;\n \n+import java.io.File;\n+import java.io.FileNotFoundException;\n+import java.io.FileReader;\n+import java.util.ArrayList;\n import java.util.Arrays;\n+import java.util.List;\n+import java.util.Queue;\n \n import junit.framework.TestCase;\n+import voldemort.ServerTestUtils;\n import voldemort.client.rebalance.RebalanceStealInfo;\n+import voldemort.cluster.Cluster;\n+import voldemort.cluster.Node;\n+import voldemort.store.StoreDefinition;\n+import voldemort.xml.StoreDefinitionsMapper;\n \n public class RebalanceUtilsTest extends TestCase {\n \n+    private static String testStoreName = "test-replication-memory";\n+    private static String storeDefFile = "test/common/voldemort/config/stores.xml";\n+\n+    private Cluster currentCluster;\n+    private Cluster targetCluster;\n+    List<StoreDefinition> storeDefList;\n+\n+    @Override\n+    public void setUp() {\n+        currentCluster = ServerTestUtils.getLocalCluster(2, new int[][] {\n+                { 0, 1, 2, 3, 4, 5, 6, 7, 8 }, {}, {}, {} });\n+\n+        targetCluster = ServerTestUtils.getLocalCluster(2, new int[][] { { 0, 1, 4, 5, 6, 7, 8 },\n+                { 2, 3 } });\n+\n+        try {\n+            storeDefList = new StoreDefinitionsMapper().readStoreList(new FileReader(new File(storeDefFile)));\n+        } catch(FileNotFoundException e) {\n+            throw new RuntimeException("Failed to find storeDefFile:" + storeDefFile, e);\n+        }\n+    }\n+\n+    public void testRebalancePlan() {\n+        Queue<Pair<Integer, List<RebalanceStealInfo>>> rebalancePlan = RebalanceUtils.getRebalanceTaskQueue(currentCluster,\n+                                                                                                            targetCluster,\n+                                                                                                            Arrays.asList(testStoreName));\n+        int[][] stealList = { {}, { 2, 3 } };\n+\n+        // the rebalancing plan should have exactly 3 entries.\n+        assertEquals("There should be three node rebalancing", 1, rebalancePlan.size());\n+        for(Pair<Integer, List<RebalanceStealInfo>> rebalanceInfo: rebalancePlan) {\n+            assertEquals("rebalanceInfo should have exactly one item", 1, rebalanceInfo.getSecond()\n+                                                                                       .size());\n+            RebalanceStealInfo expected = new RebalanceStealInfo(rebalanceInfo.getFirst(),\n+                                                                 0,\n+                                                                 listFromArray(stealList[rebalanceInfo.getFirst()]),\n+                                                                 Arrays.asList(testStoreName),\n+                                                                 0);\n+\n+            assertEquals("rebalanceStealInfo should match",\n+                         expected.toJsonString(),\n+                         rebalanceInfo.getSecond().get(0).toJsonString());\n+        }\n+    }\n+\n+    public void testUpdateCluster() {\n+        Cluster updatedCluster = RebalanceUtils.updateCluster(currentCluster,\n+                                                              new ArrayList<Node>(targetCluster.getNodes()));\n+        assertEquals("updated cluster should match targetCluster", updatedCluster, targetCluster);\n+    }\n+\n+    private List<Integer> listFromArray(int[] array) {\n+        List<Integer> list = new ArrayList<Integer>();\n+        for(int x: array) {\n+            list.add(x);\n+        }\n+\n+        return list;\n+    }\n+\n     public void testRebalanceStealInfo() {\n         RebalanceStealInfo info = new RebalanceStealInfo(0,\n                                                          1,\n', 'commit_subject': 'Rebalance unit test start passing.', 'repo_name': 'voldemort/voldemort', 'committer': {'name': 'Bhupesh Bansal', 'email': '4e893ec3dba2df1222c909030ba756a1cbb308a4@gmail.com', 'time_sec': '1261090901', 'tz_offset': '-480', 'date': {'seconds': '1261090901'}}, 'commit_tree': '40990137d732626accd5d1c55f7e1acc105db29c', 'commit_message': 'Rebalance unit test start passing.\n', 'language_name': 'PHP', 'license': 'apache-2.0', 'hunks': '@@ -16,13 +16,84 @@\n \n package voldemort.utils;\n \n+import java.io.File;\n+import java.io.FileNotFoundException;\n+import java.io.FileReader;\n+import java.util.ArrayList;\n import java.util.Arrays;\n+import java.util.List;\n+import java.util.Queue;\n \n import junit.framework.TestCase;\n+import voldemort.ServerTestUtils;\n import voldemort.client.rebalance.RebalanceStealInfo;\n+import voldemort.cluster.Cluster;\n+import voldemort.cluster.Node;\n+import voldemort.store.StoreDefinition;\n+import voldemort.xml.StoreDefinitionsMapper;\n \n public class RebalanceUtilsTest extends TestCase {\n \n+    private static String testStoreName = "test-replication-memory";\n+    private static String storeDefFile = "test/common/voldemort/config/stores.xml";\n+\n+    private Cluster currentCluster;\n+    private Cluster targetCluster;\n+    List<StoreDefinition> storeDefList;\n+\n+    @Override\n+    public void setUp() {\n+        currentCluster = ServerTestUtils.getLocalCluster(2, new int[][] {\n+                { 0, 1, 2, 3, 4, 5, 6, 7, 8 }, {}, {}, {} });\n+\n+        targetCluster = ServerTestUtils.getLocalCluster(2, new int[][] { { 0, 1, 4, 5, 6, 7, 8 },\n+                { 2, 3 } });\n+\n+        try {\n+            storeDefList = new StoreDefinitionsMapper().readStoreList(new FileReader(new File(storeDefFile)));\n+        } catch(FileNotFoundException e) {\n+            throw new RuntimeException("Failed to find storeDefFile:" + storeDefFile, e);\n+        }\n+    }\n+\n+    public void testRebalancePlan() {\n+        Queue<Pair<Integer, List<RebalanceStealInfo>>> rebalancePlan = RebalanceUtils.getRebalanceTaskQueue(currentCluster,\n+                                                                                                            targetCluster,\n+                                                                                                            Arrays.asList(testStoreName));\n+        int[][] stealList = { {}, { 2, 3 } };\n+\n+        // the rebalancing plan should have exactly 3 entries.\n+        assertEquals("There should be three node rebalancing", 1, rebalancePlan.size());\n+        for(Pair<Integer, List<RebalanceStealInfo>> rebalanceInfo: rebalancePlan) {\n+            assertEquals("rebalanceInfo should have exactly one item", 1, rebalanceInfo.getSecond()\n+                                                                                       .size());\n+            RebalanceStealInfo expected = new RebalanceStealInfo(rebalanceInfo.getFirst(),\n+                                                                 0,\n+                                                                 listFromArray(stealList[rebalanceInfo.getFirst()]),\n+                                                                 Arrays.asList(testStoreName),\n+                                                                 0);\n+\n+            assertEquals("rebalanceStealInfo should match",\n+                         expected.toJsonString(),\n+                         rebalanceInfo.getSecond().get(0).toJsonString());\n+        }\n+    }\n+\n+    public void testUpdateCluster() {\n+        Cluster updatedCluster = RebalanceUtils.updateCluster(currentCluster,\n+                                                              new ArrayList<Node>(targetCluster.getNodes()));\n+        assertEquals("updated cluster should match targetCluster", updatedCluster, targetCluster);\n+    }\n+\n+    private List<Integer> listFromArray(int[] array) {\n+        List<Integer> list = new ArrayList<Integer>();\n+        for(int x: array) {\n+            list.add(x);\n+        }\n+\n+        return list;\n+    }\n+\n     public void testRebalanceStealInfo() {\n         RebalanceStealInfo info = new RebalanceStealInfo(0,\n                                                          1,\n', 'addition_count': 71, 'deletion_count': 0, 'src_file': 'a/test/unit/voldemort/utils/RebalanceUtilsTest.java', 'tgt_file': 'b/test/unit/voldemort/utils/RebalanceUtilsTest.java', 'file_extension': '.java', 'before_file': '/*\n * Copyright 2011-2013 LinkedIn, Inc\n * \n * Licensed under the Apache License, Version 2.0 (the "License"); you may not\n * use this file except in compliance with the License. You may obtain a copy of\n * the License at\n * \n * http://www.apache.org/licenses/LICENSE-2.0\n * \n * Unless required by applicable law or agreed to in writing, software\n * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT\n * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n * License for the specific language governing permissions and limitations under\n * the License.\n */\n\npackage voldemort.utils;\n\nimport java.util.Arrays;\n\nimport junit.framework.TestCase;\nimport voldemort.client.rebalance.RebalanceStealInfo;\n\npublic class RebalanceUtilsTest extends TestCase {\n\n    public void testRebalanceStealInfo() {\n        RebalanceStealInfo info = new RebalanceStealInfo(0,\n                                                         1,\n        if(verify)\n            assertEquals(interimC, derivedInterim2);\n\n        RebalanceUtils.validateCurrentFinalCluster(currentC, finalC);\n        RebalanceUtils.validateCurrentInterimCluster(currentC, interimC);\n        RebalanceUtils.validateInterimFinalCluster(interimC, finalC);\n    }\n\n    private void doClusterTransformation(Cluster currentC, Cluster interimC, Cluster finalC) {\n        doClusterTransformationBase(currentC, interimC, finalC, false);\n    }\n\n    public void doClusterTransformationAndVerification(Cluster currentC,\n                                                       Cluster interimC,\n                                                       Cluster finalC) {\n        doClusterTransformationBase(currentC, interimC, finalC, true);\n    }\n\n    @Test\n    public void testClusterTransformationAndVerification() {\n        // Two-zone cluster: no-op\n        doClusterTransformationAndVerification(ClusterTestUtils.getZZCluster(),\n                                               ClusterTestUtils.getZZCluster(),\n                                               ClusterTestUtils.getZZCluster());\n\n        // Two-zone cluster: rebalance\n        doClusterTransformationAndVerification(ClusterTestUtils.getZZCluster(),\n                                               ClusterTestUtils.getZZCluster(),\n                                               ClusterTestUtils.getZZClusterWithSwappedPartitions());\n\n        // Two-zone cluster: cluster expansion\n        doClusterTransformationAndVerification(ClusterTestUtils.getZZCluster(),\n                                               ClusterTestUtils.getZZClusterWithNN(),\n                                               ClusterTestUtils.getZZClusterWithPP());\n\n        // Three-zone cluster: no-op\n        doClusterTransformationAndVerification(ClusterTestUtils.getZZZCluster(),\n                                               ClusterTestUtils.getZZZCluster(),\n                                               ClusterTestUtils.getZZZCluster());\n\n        // Three-zone cluster: rebalance\n        doClusterTransformationAndVerification(ClusterTestUtils.getZZZCluster(),\n                                               ClusterTestUtils.getZZZCluster(),\n                                               ClusterTestUtils.getZZZClusterWithSwappedPartitions());\n\n        // Three-zone cluster: cluster expansion\n        doClusterTransformationAndVerification(ClusterTestUtils.getZZZCluster(),\n                                               ClusterTestUtils.getZZZClusterWithNNN(),\n                                               ClusterTestUtils.getZZZClusterWithPPP());\n\n        doClusterTransformationAndVerification(ClusterTestUtils.getZZCluster(),\n                                               ClusterTestUtils.getZZECluster(),\n                                               ClusterTestUtils.getZZEClusterXXP());\n    }\n\n    @Test\n    public void testClusterTransformationAndVerificationExceptions() {\n        boolean excepted;\n\n        // Two-zone cluster: rebalance with extra partitions in interim cluster\n        excepted = false;\n        try {\n            doClusterTransformation(ClusterTestUtils.getZZCluster(),\n                                    ClusterTestUtils.getZZClusterWithExtraPartitions(),\n                                    ClusterTestUtils.getZZClusterWithSwappedPartitions());\n        } catch(VoldemortException ve) {\n            excepted = true;\n        }\n        assertTrue(excepted);\n\n        // Two-zone cluster: rebalance with extra partitions in final\n        excepted = false;\n        try {\n            doClusterTransformation(ClusterTestUtils.getZZCluster(),\n                                    ClusterTestUtils.getZZCluster(),\n                                    ClusterTestUtils.getZZClusterWithExtraPartitions());\n        } catch(VoldemortException ve) {\n            excepted = true;\n        }\n        assertTrue(excepted);\n\n        // Two-zone cluster: node ids swapped in interim cluster\n        excepted = false;\n        try {\n            doClusterTransformation(ClusterTestUtils.getZZCluster(),\n                                    ClusterTestUtils.getZZClusterWithNNWithSwappedNodeIds(),\n                                    ClusterTestUtils.getZZClusterWithPP());\n        } catch(VoldemortException ve) {\n            excepted = true;\n        }\n        assertTrue(excepted);\n\n        // Two-zone cluster: node ids swapped in final is OK because this is the\n        // same as partitions being migrated among nodes.\n        excepted = false;\n        try {\n            doClusterTransformation(ClusterTestUtils.getZZCluster(),\n                                    ClusterTestUtils.getZZClusterWithNN(),\n                                    ClusterTestUtils.getZZClusterWithPPWithSwappedNodeIds());\n        } catch(VoldemortException ve) {\n            excepted = true;\n        }\n        assertFalse(excepted);\n\n        // Two-zone cluster: too many node ids in final\n        excepted = false;\n        try {\n            doClusterTransformation(ClusterTestUtils.getZZCluster(),\n                                    ClusterTestUtils.getZZClusterWithNN(),\n                                    ClusterTestUtils.getZZClusterWithPPWithTooManyNodes());\n        } catch(VoldemortException ve) {\n            excepted = true;\n        }\n        assertTrue(excepted);\n    }\n}\n', 'message': 'Rebalance unit test start passing.\n', 'author_name': 'Bhupesh Bansal', 'lang': 'java'}