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;
        }

    }
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 + " ]" ) ; List < RebalanceTaskInfo > completedRebalanceTaskInfo = Lists . newArrayList ( ) ; List < String > swappedStoreNames = Lists . newArrayList ( ) ; boolean completedClusterAndStoresChange = false ; boolean completedRebalanceSourceClusterChange = false ; Cluster previousRebalancingSourceCluster = null ; List < StoreDefinition > previousRebalancingSourceStores = null ; try { if ( changeRebalanceState ) { try { previousRebalancingSourceCluster = metadataStore . getRebalancingSourceCluster ( ) ; previousRebalancingSourceStores = metadataStore . getRebalancingSourceStores ( ) ; if ( ! rollback ) { changeClusterAndStores ( MetadataStore . REBALANCING_SOURCE_CLUSTER_XML , currentCluster , MetadataStore . REBALANCING_SOURCE_STORES_XML , currentStoreDefs ) ; completedRebalanceSourceClusterChange = true ; for ( RebalanceTaskInfo info : rebalanceTaskInfo ) { metadataStore . addRebalancingState ( info ) ; completedRebalanceTaskInfo . add ( info ) ; } } else { changeClusterAndStores ( MetadataStore . REBALANCING_SOURCE_CLUSTER_XML , 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 ) ; } } 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 ; } if ( swapRO ) { swapROStores ( swappedStoreNames , false ) ; } } catch ( VoldemortException e ) { logger . error ( "Got exception while changing state, now rolling back changes" , e ) ; 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 ) ; } } if ( swappedStoreNames . size ( ) > 0 ) { try { swapROStores ( swappedStoreNames , true ) ; } catch ( Exception exception ) { logger . error ( "Error while swapping back to old state " , exception ) ; } } 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 ) ; } } } } 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 ; } }
Support four different stages <br>
For normal operation:

<pre>
| swapRO | changeClusterMetadata | changeRebalanceState | Order |
| f | t | t | rebalance -> cluster  |
| f | f | t | rebalance |
| t | t | f | cluster -> swap |
| t | t | t | rebalance -> cluster -> swap|
</pre>

In general we need to do [ cluster change -> swap -> rebalance state
change ]

NOTE: The update of the cluster metadata and the rebalancer state is not
"atomic". Ergo, there could theoretically be a race where a client picks
up new cluster metadata sends a request based on that, but the proxy
bridges have not been setup and we either miss a proxy put or return a
null for get/getalls

TODO:refactor The rollback logic here is too convoluted. Specifically,
the independent updates to each key could be split up into their own
methods.

@param cluster Cluster metadata to change
@param rebalanceTaskInfo List of rebalance partitions info
@param swapRO Boolean to indicate swapping of RO store
@param changeClusterAndStoresMetadata Boolean to indicate a change of
cluster metadata
@param changeRebalanceState Boolean to indicate a change in rebalance
state
@param rollback Boolean to indicate that we are rolling back or not