docs: private InboundTransferTask addTransfer(Address source, IntSet segmentsFromSource) {
        final InboundTransferTask inboundTransfer;
  
        synchronized (transferMapsLock) {
           if (trace) {
              log.tracef("Adding transfer from %s for segments %s", source, segmentsFromSource);
           }
           segmentsFromSource.removeAll(transfersBySegment.keySet());  // already in progress segments are excluded
           if (segmentsFromSource.isEmpty()) {
              if (trace) {
                 log.tracef("All segments are already in progress, skipping");
              }
              return null;
           }
  
           inboundTransfer = new InboundTransferTask(segmentsFromSource, source, cacheTopology.getTopologyId(),
                 rpcManager, commandsFactory, timeout, cacheName, true);
           addTransfer(inboundTransfer, segmentsFromSource);
        }
  
        stateRequestExecutor.executeAsync(() -> {
           CompletableFuture<Void> transferStarted = inboundTransfer.requestSegments();
  
           if (trace)
              log.tracef("Waiting for inbound transfer to finish: %s", inboundTransfer);
           return transferStarted.whenComplete((aVoid, throwable) -> onTaskCompletion(inboundTransfer));
        });
        return inboundTransfer;
     }