src/java/org/apache/cassandra/service/StorageService.java
// public & static for testing purposes
public & static for testing purposes
public & static for testing purposes
public static void calculatePendingRanges(AbstractReplicationStrategy strategy, String table)
    {
        TokenMetadata tm = StorageService.instance.getTokenMetadata();
        Multimap<Range<Token>, InetAddress> pendingRanges = HashMultimap.create();
        Map<Token, InetAddress> bootstrapTokens = tm.getBootstrapTokens();
        Set<InetAddress> leavingEndpoints = tm.getLeavingEndpoints();
        if (bootstrapTokens.isEmpty() && leavingEndpoints.isEmpty() && tm.getMovingEndpoints().isEmpty())
        {
            if (logger_.isDebugEnabled())
                logger_.debug("No bootstrapping, leaving or moving nodes -> empty pending ranges for {}", table);
            tm.setPendingRanges(table, pendingRanges);
            return;
        }
        Multimap<InetAddress, Range<Token>> addressRanges = strategy.getAddressRanges();
        TokenMetadata allLeftMetadata = tm.cloneAfterAllLeft();
        Set<Range<Token>> affectedRanges = new HashSet<Range<Token>>();
        for (InetAddress endpoint : leavingEndpoints)
            affectedRanges.addAll(addressRanges.get(endpoint));
        for (Range<Token> range : affectedRanges)
        {
            Set<InetAddress> currentEndpoints = ImmutableSet.copyOf(strategy.calculateNaturalEndpoints(range.right, tm));
            Set<InetAddress> newEndpoints = ImmutableSet.copyOf(strategy.calculateNaturalEndpoints(range.right, allLeftMetadata));
            pendingRanges.putAll(range, Sets.difference(newEndpoints, currentEndpoints));
        }
        for (Map.Entry<Token, InetAddress> entry : bootstrapTokens.entrySet())
        {
            InetAddress endpoint = entry.getValue();
            allLeftMetadata.updateNormalToken(entry.getKey(), endpoint);
            for (Range<Token> range : strategy.getAddressRanges(allLeftMetadata).get(endpoint))
                pendingRanges.put(range, endpoint);
            allLeftMetadata.removeEndpoint(endpoint);
        }
        for (Pair<Token, InetAddress> moving : tm.getMovingEndpoints())
        {
            InetAddress endpoint = moving.right; 
            allLeftMetadata.updateNormalToken(moving.left, endpoint);
            for (Range<Token> range : strategy.getAddressRanges(allLeftMetadata).get(endpoint))
            {
                pendingRanges.put(range, endpoint);
            }
            allLeftMetadata.removeEndpoint(endpoint);
        }
        tm.setPendingRanges(table, pendingRanges);
        if (logger_.isDebugEnabled())
            logger_.debug("Pending ranges:\n" + (pendingRanges.isEmpty() ? "<empty>" : tm.printPendingRanges()));
    }
public static void calculatePendingRanges ( AbstractReplicationStrategy strategy , String table ) { TokenMetadata tm = StorageService . instance . getTokenMetadata ( ) ; Multimap < Range < Token > , InetAddress > pendingRanges = HashMultimap . create ( ) ; Map < Token , InetAddress > bootstrapTokens = tm . getBootstrapTokens ( ) ; Set < InetAddress > leavingEndpoints = tm . getLeavingEndpoints ( ) ; if ( bootstrapTokens . isEmpty ( ) && leavingEndpoints . isEmpty ( ) && tm . getMovingEndpoints ( ) . isEmpty ( ) ) { if ( logger_ . isDebugEnabled ( ) ) logger_ . debug ( "No bootstrapping, leaving or moving nodes -> empty pending ranges for {}" , table ) ; tm . setPendingRanges ( table , pendingRanges ) ; return ; } Multimap < InetAddress , Range < Token > > addressRanges = strategy . getAddressRanges ( ) ; TokenMetadata allLeftMetadata = tm . cloneAfterAllLeft ( ) ; Set < Range < Token > > affectedRanges = new HashSet < Range < Token > > ( ) ; for ( InetAddress endpoint : leavingEndpoints ) affectedRanges . addAll ( addressRanges . get ( endpoint ) ) ; for ( Range < Token > range : affectedRanges ) { Set < InetAddress > currentEndpoints = ImmutableSet . copyOf ( strategy . calculateNaturalEndpoints ( range . right , tm ) ) ; Set < InetAddress > newEndpoints = ImmutableSet . copyOf ( strategy . calculateNaturalEndpoints ( range . right , allLeftMetadata ) ) ; pendingRanges . putAll ( range , Sets . difference ( newEndpoints , currentEndpoints ) ) ; } for ( Map . Entry < Token , InetAddress > entry : bootstrapTokens . entrySet ( ) ) { InetAddress endpoint = entry . getValue ( ) ; allLeftMetadata . updateNormalToken ( entry . getKey ( ) , endpoint ) ; for ( Range < Token > range : strategy . getAddressRanges ( allLeftMetadata ) . get ( endpoint ) ) pendingRanges . put ( range , endpoint ) ; allLeftMetadata . removeEndpoint ( endpoint ) ; } for ( Pair < Token , InetAddress > moving : tm . getMovingEndpoints ( ) ) { InetAddress endpoint = moving . right ; allLeftMetadata . updateNormalToken ( moving . left , endpoint ) ; for ( Range < Token > range : strategy . getAddressRanges ( allLeftMetadata ) . get ( endpoint ) ) { pendingRanges . put ( range , endpoint ) ; } allLeftMetadata . removeEndpoint ( endpoint ) ; } tm . setPendingRanges ( table , pendingRanges ) ; if ( logger_ . isDebugEnabled ( ) ) logger_ . debug ( "Pending ranges:\n" + ( pendingRanges . isEmpty ( ) ? "<empty>" : tm . printPendingRanges ( ) ) ) ; }
public & static for testing purposes
public & static for testing purposes
// Copy of metadata reflecting the situation after all leave operations are finished. // get all ranges that will be affected by leaving nodes // for each of those ranges, find what new nodes will be responsible for the range when // all leaving nodes are gone. // At this stage pendingRanges has been updated according to leave operations. We can // now continue the calculation by checking bootstrapping nodes. // For each of the bootstrapping nodes, simply add and remove them one by one to // allLeftMetadata and check in between what their ranges would be. // At this stage pendingRanges has been updated according to leaving and bootstrapping nodes. // We can now finish the calculation by checking moving nodes. // For each of the moving nodes, we do the same thing we did for bootstrapping: // simply add and remove them one by one to allLeftMetadata and check in between what their ranges would be. // address of the moving node //  moving.left is a new token of the endpoint