docs: public void rebalanceStateChange(Cluster cluster,
                                       List<StoreDefinition> storeDefs,
                                       List<RebalanceTaskInfo> rebalanceTaskInfo,
                                       boolean swapRO,
                                       boolean changeClusterAndStoresMetadata,
                                       boolean changeRebalanceState,
                                       boolean rollback) {
          Cluster currentCluster = metadataStore.getCluster();
          List<StoreDefinition> currentStoreDefs = metadataStore.getStoreDefList();
  
          logger.info("Server doing rebalance state change with options [ cluster metadata change - "
                      + changeClusterAndStoresMetadata + " ], [ changing rebalancing state - "
                      + changeRebalanceState + " ], [ changing swapping RO - " + swapRO
                      + " ], [ rollback - " + rollback + " ]");
  
          // Variables to track what has completed
          List<RebalanceTaskInfo> completedRebalanceTaskInfo = Lists.newArrayList();
          List<String> swappedStoreNames = Lists.newArrayList();
          boolean completedClusterAndStoresChange = false;
          boolean completedRebalanceSourceClusterChange = false;
          Cluster previousRebalancingSourceCluster = null;
          List<StoreDefinition> previousRebalancingSourceStores = null;
  
          try {
  
              /*
               * Do the rebalancing state changes. It is important that this
               * happens before the actual cluster metadata is changed. Here's
               * what could happen otherwise. When a batch completes with
               * {current_cluster c2, rebalancing_source_cluster c1} and the next
               * rebalancing state changes it to {current_cluster c3,
               * rebalancing_source_cluster c2} is set for the next batch, then
               * there could be a window during which the state is
               * {current_cluster c3, rebalancing_source_cluster c1}. On the other
               * hand, when we update the rebalancing source cluster first, there
               * is a window where the state is {current_cluster c2,
               * rebalancing_source_cluster c2}, which still fine, because of the
               * following. Successful completion of a batch means the cluster is
               * finalized, so its okay to stop proxying based on {current_cluster
               * c2, rebalancing_source_cluster c1}. And since the cluster
               * metadata has not yet been updated to c3, the writes will happen
               * based on c2.
               * 
               * 
               * Even if some clients have already seen the {current_cluster c3,
               * rebalancing_source_cluster c2} state from other servers, the
               * operation will be rejected with InvalidMetadataException since
               * this server itself is not aware of C3
               */
  
              // CHANGE REBALANCING STATE
              if(changeRebalanceState) {
                  try {
                      previousRebalancingSourceCluster = metadataStore.getRebalancingSourceCluster();
                      previousRebalancingSourceStores = metadataStore.getRebalancingSourceStores();
                      if(!rollback) {
  
                          // Save up the current cluster and stores def for
                          // Redirecting store
                          changeClusterAndStores(MetadataStore.REBALANCING_SOURCE_CLUSTER_XML,
                                                 currentCluster,
                                                 // Save the current store defs
                                                 // for Redirecting store
                                                 MetadataStore.REBALANCING_SOURCE_STORES_XML,
                                                 currentStoreDefs);
  
                          completedRebalanceSourceClusterChange = true;
  
                          for(RebalanceTaskInfo info: rebalanceTaskInfo) {
                              metadataStore.addRebalancingState(info);
                              completedRebalanceTaskInfo.add(info);
                          }
                      } else {
                          // Reset the rebalancing source cluster back to null
  
                          changeClusterAndStores(MetadataStore.REBALANCING_SOURCE_CLUSTER_XML, null,
                                                 // Reset the rebalancing source
                                                 // stores back to null
                                                 MetadataStore.REBALANCING_SOURCE_STORES_XML,
                                                 null);
  
                          completedRebalanceSourceClusterChange = true;
  
                          for(RebalanceTaskInfo info: rebalanceTaskInfo) {
                              metadataStore.deleteRebalancingState(info);
                              completedRebalanceTaskInfo.add(info);
                          }
                      }
                  } catch(Exception e) {
                      throw new VoldemortException(e);
                  }
              }
  
              // CHANGE CLUSTER METADATA AND STORE METADATA
              if(changeClusterAndStoresMetadata) {
                  logger.info("Switching cluster metadata from " + currentCluster + " to " + cluster);
                  logger.info("Switching stores metadata from " + currentStoreDefs + " to "
                              + storeDefs);
                  changeClusterAndStores(MetadataStore.CLUSTER_KEY,
                                         cluster,
                                         MetadataStore.STORES_KEY,
                                         storeDefs);
  
                  completedClusterAndStoresChange = true;
  
              }
  
              // SWAP RO DATA FOR ALL STORES
              if(swapRO) {
                  swapROStores(swappedStoreNames, false);
              }
  
          } catch(VoldemortException e) {
  
              logger.error("Got exception while changing state, now rolling back changes", e);
  
              // ROLLBACK CLUSTER AND STORES CHANGE
              if(completedClusterAndStoresChange) {
                  try {
                      logger.info("Rolling back cluster.xml to " + currentCluster);
                      logger.info("Rolling back stores.xml to " + currentStoreDefs);
                      changeClusterAndStores(MetadataStore.CLUSTER_KEY,
                                             currentCluster,
                                             MetadataStore.STORES_KEY,
                                             currentStoreDefs);
                  } catch(Exception exception) {
                      logger.error("Error while rolling back cluster metadata to " + currentCluster
                                   + " Stores metadata to " + currentStoreDefs, exception);
                  }
              }
  
              // SWAP RO DATA FOR ALL COMPLETED STORES
              if(swappedStoreNames.size() > 0) {
                  try {
                      swapROStores(swappedStoreNames, true);
                  } catch(Exception exception) {
                      logger.error("Error while swapping back to old state ", exception);
                  }
              }
  
              // CHANGE BACK ALL REBALANCING STATES FOR COMPLETED ONES
              if(completedRebalanceTaskInfo.size() > 0) {
                  if(!rollback) {
                      for(RebalanceTaskInfo info: completedRebalanceTaskInfo) {
                          try {
                              metadataStore.deleteRebalancingState(info);
                          } catch(Exception exception) {
                              logger.error("Error while deleting back rebalance info during error rollback "
                                                   + info,
                                           exception);
                          }
                      }
                  } else {
                      for(RebalanceTaskInfo info: completedRebalanceTaskInfo) {
                          try {
                              metadataStore.addRebalancingState(info);
                          } catch(Exception exception) {
                              logger.error("Error while adding back rebalance info during error rollback "
                                                   + info,
                                           exception);
                          }
                      }
                  }
  
              }
  
              // Revert changes to REBALANCING_SOURCE_CLUSTER_XML and
              // REBALANCING_SOURCE_STORES_XML
              if(completedRebalanceSourceClusterChange) {
                  logger.info("Reverting the REBALANCING_SOURCE_CLUSTER_XML back to "
                              + previousRebalancingSourceCluster);
                  logger.info("Reverting the REBALANCING_SOURCE_STORES_XML back to "
                              + previousRebalancingSourceStores);
                  changeClusterAndStores(MetadataStore.REBALANCING_SOURCE_CLUSTER_XML,
                                         previousRebalancingSourceCluster,
                                         MetadataStore.REBALANCING_SOURCE_STORES_XML,
                                         previousRebalancingSourceStores);
              }
  
              throw e;
          }
  
      }