private void restartBrokenTransfers ( CacheTopology cacheTopology , IntSet addedSegments ) { Set < Address > members = new HashSet <> ( cacheTopology . getReadConsistentHash ( ) . getMembers ( ) ) ; synchronized ( transferMapsLock ) { for ( Iterator < Map . Entry < Address , List < InboundTransferTask > > > it = transfersBySource . entrySet ( ) . iterator ( ) ; it . hasNext ( ) ; ) { Map . Entry < Address , List < InboundTransferTask > > entry = it . next ( ) ; Address source = entry . getKey ( ) ; if ( ! members . contains ( source ) ) { if ( trace ) { log . tracef ( "Removing inbound transfers from source %s for cache %s" , source , cacheName ) ; } List < InboundTransferTask > inboundTransfers = entry . getValue ( ) ; it . remove ( ) ; for ( InboundTransferTask inboundTransfer : inboundTransfers ) { // these segments will be restarted if they are still in new write CH if ( trace ) { log . tracef ( "Removing inbound transfers from node %s for segments %s" , source , inboundTransfer . getSegments ( ) ) ; } IntSet unfinishedSegments = inboundTransfer . getUnfinishedSegments ( ) ; inboundTransfer . cancel ( ) ; addedSegments . addAll ( unfinishedSegments ) ; transfersBySegment . keySet ( ) . removeAll ( unfinishedSegments ) ; } } } // exclude those that are already in progress from a valid source addedSegments . removeAll ( transfersBySegment . keySet ( ) ) ; } }
Check if any of the existing transfers should be restarted from a different source because the initial source is no longer a member .