src/java/org/apache/cassandra/service/StorageService.java
/**
     * move the node to new token or find a new token to boot to according to load
     *
     * @param newToken new token to boot to, or if null, find balanced token to boot to
     *
     * @throws IOException on any I/O operation error
     */
move the node to new token or find a new token to boot to according to load
@param newToken new token to boot to, or if null, find balanced token to boot to

@throws IOException on any I/O operation error
move the node to new token or find a new token to boot to according to load @param newToken new token to boot to or if null find balanced token to boot to @throws IOException on any I / O operation error
private void move(Token newToken) throws IOException
    {
        if (newToken == null)
            throw new IOException("Can't move to the undefined (null) token.");
        if (tokenMetadata_.sortedTokens().contains(newToken))
            throw new IOException("target token " + newToken + " is already owned by another node.");
        InetAddress localAddress = FBUtilities.getBroadcastAddress();
        List<String> tablesToProcess = Schema.instance.getNonSystemTables();
        for (String table : tablesToProcess)
        {
            if (tokenMetadata_.getPendingRanges(table, localAddress).size() > 0)
                throw new UnsupportedOperationException("data is currently moving to this node; unable to leave the ring");
        }
        Gossiper.instance.addLocalApplicationState(ApplicationState.STATUS, valueFactory.moving(newToken));
        logger_.info(String.format("Moving %s from %s to %s.", localAddress, getLocalToken(), newToken));
        IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
        Map<String, Multimap<InetAddress, Range<Token>>> rangesToFetch = new HashMap<String, Multimap<InetAddress, Range<Token>>>();
        Map<String, Multimap<Range<Token>, InetAddress>> rangesToStreamByTable = new HashMap<String, Multimap<Range<Token>, InetAddress>>();
        TokenMetadata tokenMetaClone = tokenMetadata_.cloneAfterAllSettled();
        for (String table : tablesToProcess)
        {
            AbstractReplicationStrategy strategy = Table.open(table).getReplicationStrategy();
            Collection<Range<Token>> currentRanges = getRangesForEndpoint(table, localAddress);
            Collection<Range<Token>> updatedRanges = strategy.getPendingAddressRanges(tokenMetadata_, newToken, localAddress);
            Multimap<Range<Token>, InetAddress> rangeAddresses = strategy.getRangeAddresses(tokenMetadata_);
            Pair<Set<Range<Token>>, Set<Range<Token>>> rangesPerTable = calculateStreamAndFetchRanges(currentRanges, updatedRanges);
            /**
             * In this loop we are going through all ranges "to fetch" and determining
             * nodes in the ring responsible for data we are interested in
             */
            Multimap<Range<Token>, InetAddress> rangesToFetchWithPreferredEndpoints = ArrayListMultimap.create();
            for (Range<Token> toFetch : rangesPerTable.right)
            {
                for (Range<Token> range : rangeAddresses.keySet())
                {
                    if (range.contains(toFetch))
                    {
                        List<InetAddress> endpoints = snitch.getSortedListByProximity(localAddress, rangeAddresses.get(range));
                        rangesToFetchWithPreferredEndpoints.putAll(toFetch, endpoints);
                    }
                }
            }
            Multimap<Range<Token>, InetAddress> rangeWithEndpoints = HashMultimap.create();
            for (Range<Token> toStream : rangesPerTable.left)
            {
                Set<InetAddress> currentEndpoints = ImmutableSet.copyOf(strategy.calculateNaturalEndpoints(toStream.right, tokenMetadata_));
                Set<InetAddress> newEndpoints = ImmutableSet.copyOf(strategy.calculateNaturalEndpoints(toStream.right, tokenMetaClone));
                rangeWithEndpoints.putAll(toStream, Sets.difference(newEndpoints, currentEndpoints));
            }
            rangesToStreamByTable.put(table, rangeWithEndpoints);
            Multimap<InetAddress, Range<Token>> workMap = BootStrapper.getWorkMap(rangesToFetchWithPreferredEndpoints);
            rangesToFetch.put(table, workMap);
            if (logger_.isDebugEnabled())
                logger_.debug("Table {}: work map {}.", table, workMap);
        }
        if (!rangesToStreamByTable.isEmpty() || !rangesToFetch.isEmpty())
        {
            logger_.info("Sleeping {} ms before start streaming/fetching ranges.", RING_DELAY);
            try
            {
                Thread.sleep(RING_DELAY);
            }
            catch (InterruptedException e)
            {
                throw new RuntimeException("Sleep interrupted " + e.getMessage());
            }
            setMode(Mode.MOVING, "fetching new ranges and streaming old ranges", true);
            if (logger_.isDebugEnabled())
                logger_.debug("[Move->STREAMING] Work Map: " + rangesToStreamByTable);
            CountDownLatch streamLatch = streamRanges(rangesToStreamByTable);
            if (logger_.isDebugEnabled())
                logger_.debug("[Move->FETCHING] Work Map: " + rangesToFetch);
            CountDownLatch fetchLatch = requestRanges(rangesToFetch);
            try
            {
                streamLatch.await();
                fetchLatch.await();
            }
            catch (InterruptedException e)
            {
                throw new RuntimeException("Interrupted latch while waiting for stream/fetch ranges to finish: " + e.getMessage());
            }
        }
        setToken(newToken); 
        if (logger_.isDebugEnabled())
            logger_.debug("Successfully moved to new token {}", getLocalToken());
    }
private void move ( Token newToken ) throws IOException { if ( newToken == null ) throw new IOException ( "Can't move to the undefined (null) token." ) ; if ( tokenMetadata_ . sortedTokens ( ) . contains ( newToken ) ) throw new IOException ( "target token " + newToken + " is already owned by another node." ) ; InetAddress localAddress = FBUtilities . getBroadcastAddress ( ) ; List < String > tablesToProcess = Schema . instance . getNonSystemTables ( ) ; for ( String table : tablesToProcess ) { if ( tokenMetadata_ . getPendingRanges ( table , localAddress ) . size ( ) > 0 ) throw new UnsupportedOperationException ( "data is currently moving to this node; unable to leave the ring" ) ; } Gossiper . instance . addLocalApplicationState ( ApplicationState . STATUS , valueFactory . moving ( newToken ) ) ; logger_ . info ( String . format ( "Moving %s from %s to %s." , localAddress , getLocalToken ( ) , newToken ) ) ; IEndpointSnitch snitch = DatabaseDescriptor . getEndpointSnitch ( ) ; Map < String , Multimap < InetAddress , Range < Token > > > rangesToFetch = new HashMap < String , Multimap < InetAddress , Range < Token > > > ( ) ; Map < String , Multimap < Range < Token > , InetAddress > > rangesToStreamByTable = new HashMap < String , Multimap < Range < Token > , InetAddress > > ( ) ; TokenMetadata tokenMetaClone = tokenMetadata_ . cloneAfterAllSettled ( ) ; for ( String table : tablesToProcess ) { AbstractReplicationStrategy strategy = Table . open ( table ) . getReplicationStrategy ( ) ; Collection < Range < Token > > currentRanges = getRangesForEndpoint ( table , localAddress ) ; Collection < Range < Token > > updatedRanges = strategy . getPendingAddressRanges ( tokenMetadata_ , newToken , localAddress ) ; Multimap < Range < Token > , InetAddress > rangeAddresses = strategy . getRangeAddresses ( tokenMetadata_ ) ; Pair < Set < Range < Token > > , Set < Range < Token > > > rangesPerTable = calculateStreamAndFetchRanges ( currentRanges , updatedRanges ) ; /**
             * In this loop we are going through all ranges "to fetch" and determining
             * nodes in the ring responsible for data we are interested in
             */ Multimap < Range < Token > , InetAddress > rangesToFetchWithPreferredEndpoints = ArrayListMultimap . create ( ) ; for ( Range < Token > toFetch : rangesPerTable . right ) { for ( Range < Token > range : rangeAddresses . keySet ( ) ) { if ( range . contains ( toFetch ) ) { List < InetAddress > endpoints = snitch . getSortedListByProximity ( localAddress , rangeAddresses . get ( range ) ) ; rangesToFetchWithPreferredEndpoints . putAll ( toFetch , endpoints ) ; } } } Multimap < Range < Token > , InetAddress > rangeWithEndpoints = HashMultimap . create ( ) ; for ( Range < Token > toStream : rangesPerTable . left ) { Set < InetAddress > currentEndpoints = ImmutableSet . copyOf ( strategy . calculateNaturalEndpoints ( toStream . right , tokenMetadata_ ) ) ; Set < InetAddress > newEndpoints = ImmutableSet . copyOf ( strategy . calculateNaturalEndpoints ( toStream . right , tokenMetaClone ) ) ; rangeWithEndpoints . putAll ( toStream , Sets . difference ( newEndpoints , currentEndpoints ) ) ; } rangesToStreamByTable . put ( table , rangeWithEndpoints ) ; Multimap < InetAddress , Range < Token > > workMap = BootStrapper . getWorkMap ( rangesToFetchWithPreferredEndpoints ) ; rangesToFetch . put ( table , workMap ) ; if ( logger_ . isDebugEnabled ( ) ) logger_ . debug ( "Table {}: work map {}." , table , workMap ) ; } if ( ! rangesToStreamByTable . isEmpty ( ) || ! rangesToFetch . isEmpty ( ) ) { logger_ . info ( "Sleeping {} ms before start streaming/fetching ranges." , RING_DELAY ) ; try { Thread . sleep ( RING_DELAY ) ; } catch ( InterruptedException e ) { throw new RuntimeException ( "Sleep interrupted " + e . getMessage ( ) ) ; } setMode ( Mode . MOVING , "fetching new ranges and streaming old ranges" , true ) ; if ( logger_ . isDebugEnabled ( ) ) logger_ . debug ( "[Move->STREAMING] Work Map: " + rangesToStreamByTable ) ; CountDownLatch streamLatch = streamRanges ( rangesToStreamByTable ) ; if ( logger_ . isDebugEnabled ( ) ) logger_ . debug ( "[Move->FETCHING] Work Map: " + rangesToFetch ) ; CountDownLatch fetchLatch = requestRanges ( rangesToFetch ) ; try { streamLatch . await ( ) ; fetchLatch . await ( ) ; } catch ( InterruptedException e ) { throw new RuntimeException ( "Interrupted latch while waiting for stream/fetch ranges to finish: " + e . getMessage ( ) ) ; } } setToken ( newToken ) ; if ( logger_ . isDebugEnabled ( ) ) logger_ . debug ( "Successfully moved to new token {}" , getLocalToken ( ) ) ; }
move the node to new token or find a new token to boot to according to load
 @param newToken new token to boot to, or if null, find balanced token to boot to
move the node to new token or find a new token to boot to according to load @param newToken new token to boot to or if null find balanced token to boot to
// address of the current node // checking if data is moving to this node // setting 'moving' application state // for each of the non system tables calculating new ranges // which current node will handle after move to the new token // replication strategy of the current keyspace (aka table) // getting collection of the currently used ranges by this keyspace // collection of ranges which this node will serve after move to the new token // ring ranges and endpoints associated with them // this used to determine what nodes should we ping about range data // calculated parts of the ranges to request/stream from/to nodes in the ring // storing range and preferred endpoint set // calculating endpoints to stream current ranges to if needed // in some situations node will handle current ranges as part of the new ranges // associating table with range-to-endpoints map // setting new token as we have everything settled