src/java/org/apache/cassandra/service/StorageService.java
/**
     * Called when an endpoint is removed from the ring. This function checks
     * whether this node becomes responsible for new ranges as a
     * consequence and streams data if needed.
     *
     * This is rather ineffective, but it does not matter so much
     * since this is called very seldom
     *
     * @param endpoint the node that left
     */
Called when an endpoint is removed from the ring. This function checks
whether this node becomes responsible for new ranges as a
consequence and streams data if needed.

This is rather ineffective, but it does not matter so much
since this is called very seldom

@param endpoint the node that left
Called when an endpoint is removed from the ring . This function checks whether this node becomes responsible for new ranges as a consequence and streams data if needed . This is rather ineffective but it does not matter so much since this is called very seldom @param endpoint the node that left
private void restoreReplicaCount(InetAddress endpoint, final InetAddress notifyEndpoint)
    {
        final Multimap<InetAddress, String> fetchSources = HashMultimap.create();
        Multimap<String, Map.Entry<InetAddress, Collection<Range<Token>>>> rangesToFetch = HashMultimap.create();
        final InetAddress myAddress = FBUtilities.getBroadcastAddress();
        for (String table : Schema.instance.getNonSystemTables())
        {
            Multimap<Range<Token>, InetAddress> changedRanges = getChangedRangesForLeaving(table, endpoint);
            Set<Range<Token>> myNewRanges = new HashSet<Range<Token>>();
            for (Map.Entry<Range<Token>, InetAddress> entry : changedRanges.entries())
            {
                if (entry.getValue().equals(myAddress))
                    myNewRanges.add(entry.getKey());
            }
            Multimap<InetAddress, Range<Token>> sourceRanges = getNewSourceRanges(table, myNewRanges);
            for (Map.Entry<InetAddress, Collection<Range<Token>>> entry : sourceRanges.asMap().entrySet())
            {
                fetchSources.put(entry.getKey(), table);
                rangesToFetch.put(table, entry);
            }
        }
        for (final String table : rangesToFetch.keySet())
        {
            for (Map.Entry<InetAddress, Collection<Range<Token>>> entry : rangesToFetch.get(table))
            {
                final InetAddress source = entry.getKey();
                Collection<Range<Token>> ranges = entry.getValue();
                final Runnable callback = new Runnable()
                {
                    @Override
                    public void run()
                    {
                        synchronized (fetchSources)
                        {
                            fetchSources.remove(source, table);
                            if (fetchSources.isEmpty())
                                sendReplicationNotification(myAddress, notifyEndpoint);
                        }
                    }
                };
                if (logger_.isDebugEnabled())
                    logger_.debug("Requesting from " + source + " ranges " + StringUtils.join(ranges, ", "));
                StreamIn.requestRanges(source, table, ranges, callback, OperationType.RESTORE_REPLICA_COUNT);
            }
        }
    }
private void restoreReplicaCount ( InetAddress endpoint , final InetAddress notifyEndpoint ) { final Multimap < InetAddress , String > fetchSources = HashMultimap . create ( ) ; Multimap < String , Map . Entry < InetAddress , Collection < Range < Token > > > > rangesToFetch = HashMultimap . create ( ) ; final InetAddress myAddress = FBUtilities . getBroadcastAddress ( ) ; for ( String table : Schema . instance . getNonSystemTables ( ) ) { Multimap < Range < Token > , InetAddress > changedRanges = getChangedRangesForLeaving ( table , endpoint ) ; Set < Range < Token > > myNewRanges = new HashSet < Range < Token > > ( ) ; for ( Map . Entry < Range < Token > , InetAddress > entry : changedRanges . entries ( ) ) { if ( entry . getValue ( ) . equals ( myAddress ) ) myNewRanges . add ( entry . getKey ( ) ) ; } Multimap < InetAddress , Range < Token > > sourceRanges = getNewSourceRanges ( table , myNewRanges ) ; for ( Map . Entry < InetAddress , Collection < Range < Token > > > entry : sourceRanges . asMap ( ) . entrySet ( ) ) { fetchSources . put ( entry . getKey ( ) , table ) ; rangesToFetch . put ( table , entry ) ; } } for ( final String table : rangesToFetch . keySet ( ) ) { for ( Map . Entry < InetAddress , Collection < Range < Token > > > entry : rangesToFetch . get ( table ) ) { final InetAddress source = entry . getKey ( ) ; Collection < Range < Token > > ranges = entry . getValue ( ) ; final Runnable callback = new Runnable ( ) { @ Override public void run ( ) { synchronized ( fetchSources ) { fetchSources . remove ( source , table ) ; if ( fetchSources . isEmpty ( ) ) sendReplicationNotification ( myAddress , notifyEndpoint ) ; } } } ; if ( logger_ . isDebugEnabled ( ) ) logger_ . debug ( "Requesting from " + source + " ranges " + StringUtils . join ( ranges , ", " ) ) ; StreamIn . requestRanges ( source , table , ranges , callback , OperationType . RESTORE_REPLICA_COUNT ) ; } } }
Called when an endpoint is removed from the ring.
Called when an endpoint is removed from the ring .