core/src/main/java/org/infinispan/statetransfer/StateConsumerImpl.java
/**
    * Check if any of the existing transfers should be restarted from a different source because the initial source is no longer a member.
    */
Check if any of the existing transfers should be restarted from a different source because the initial source is no longer a member.
Check if any of the existing transfers should be restarted from a different source because the initial source is no longer a member .
private void restartBrokenTransfers(CacheTopology cacheTopology, Set<Integer> 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) {
                  if (trace) {
                     log.tracef("Removing inbound transfers from node %s for segments %s", source, inboundTransfer.getSegments());
                  }
                  inboundTransfer.cancel();
                  transfersBySegment.keySet().removeAll(inboundTransfer.getSegments());
                  addedSegments.addAll(inboundTransfer.getUnfinishedSegments());
               }
            }
         }
         addedSegments.removeAll(transfersBySegment.keySet());
      }
   }
private void restartBrokenTransfers ( CacheTopology cacheTopology , Set < Integer > 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 ) { if ( trace ) { log . tracef ( "Removing inbound transfers from node %s for segments %s" , source , inboundTransfer . getSegments ( ) ) ; } inboundTransfer . cancel ( ) ; transfersBySegment . keySet ( ) . removeAll ( inboundTransfer . getSegments ( ) ) ; addedSegments . addAll ( inboundTransfer . getUnfinishedSegments ( ) ) ; } } } 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.
Check if any of the existing transfers should be restarted from a different source because the initial source is no longer a member .
// these segments will be restarted if they are still in new write CH // exclude those that are already in progress from a valid source