docs: 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());
        }
     }