private void requestFragmenting(IgfsEntryInfo fileInfo){
  IgfsFileMap map=fileInfo.fileMap();
  assert map != null && !map.ranges().isEmpty();
  Map<UUID,Collection<IgfsFileAffinityRange>> grpMap=U.newHashMap(map.ranges().size());
  for (  IgfsFileAffinityRange range : map.ranges()) {
    UUID nodeId=igfsCtx.data().affinityNode(range.affinityKey()).id();
    Collection<IgfsFileAffinityRange> nodeRanges=grpMap.get(nodeId);
    if (nodeRanges == null) {
      nodeRanges=new LinkedList<>();
      grpMap.put(nodeId,nodeRanges);
    }
    nodeRanges.addAll(range.split(igfsCtx.data().groupBlockSize()));
  }
  Collection<UUID> nodeIds=new IdentityHashSet(grpMap.keySet());
  if (log.isDebugEnabled())   log.debug("Calculating fragmentizer groups for file [fileInfo=" + fileInfo + ", nodeIds="+ nodeIds+ ']');
  Object old=fragmentingFiles.putIfAbsent(fileInfo.id(),nodeIds);
  assert old == null;
  for (  Map.Entry<UUID,Collection<IgfsFileAffinityRange>> entry : grpMap.entrySet()) {
    UUID nodeId=entry.getKey();
    IgfsFragmentizerRequest msg=new IgfsFragmentizerRequest(fileInfo.id(),entry.getValue());
    try {
      if (log.isDebugEnabled())       log.debug("Sending fragmentizer request to remote node [nodeId=" + nodeId + ", fileId="+ fileInfo.id()+ ", msg="+ msg+ ']');
      sendWithRetries(nodeId,msg);
    }
 catch (    IgniteCheckedException e) {
      if (e.hasCause(ClusterTopologyCheckedException.class)) {
        if (log.isDebugEnabled())         log.debug("Failed to send fragmentizer request to remote node (node left grid): " + nodeId);
      }
 else       U.error(log,"Failed to send fragmentizer request to remote node [nodeId=" + nodeId + ", msg="+ msg+ ']',e);
      nodeIds.remove(nodeId);
    }
  }
  if (nodeIds.isEmpty()) {
    if (log.isDebugEnabled())     log.debug("Got empty wait set for fragmentized file: " + fileInfo);
    fragmentingFiles.remove(fileInfo.id(),nodeIds);
  }
}

private void requestFragmenting ( IgfsEntryInfo fileInfo ) { IgfsFileMap map = fileInfo . fileMap ( ) ; assert map != null && ! map . ranges ( ) . isEmpty ( ) ; Map < UUID , Collection < IgfsFileAffinityRange > > grpMap = U . newHashMap ( map . ranges ( ) . size ( ) ) ; for ( IgfsFileAffinityRange range : map . ranges ( ) ) { UUID nodeId = igfsCtx . data ( ) . affinityNode ( range . affinityKey ( ) ) . id ( ) ; Collection < IgfsFileAffinityRange > nodeRanges = grpMap . get ( nodeId ) ; if ( nodeRanges == null ) { nodeRanges = new LinkedList < > ( ) ; grpMap . put ( nodeId , nodeRanges ) ; } nodeRanges . addAll ( range . split ( igfsCtx . data ( ) . groupBlockSize ( ) ) ) ; } Collection < UUID > nodeIds = new IdentityHashSet ( grpMap . keySet ( ) ) ; if ( log . isDebugEnabled ( ) ) log . debug ( "Calculating fragmentizer groups for file [fileInfo=" + fileInfo + ", nodeIds=" + nodeIds + ']' ) ; Object old = fragmentingFiles . putIfAbsent ( fileInfo . id ( ) , nodeIds ) ; assert old == null ; for ( Map . Entry < UUID , Collection < IgfsFileAffinityRange > > entry : grpMap . entrySet ( ) ) { UUID nodeId = entry . getKey ( ) ; IgfsFragmentizerRequest msg = new IgfsFragmentizerRequest ( fileInfo . id ( ) , entry . getValue ( ) ) ; try { if ( log . isDebugEnabled ( ) ) log . debug ( "Sending fragmentizer request to remote node [nodeId=" + nodeId + ", fileId=" + fileInfo . id ( ) + ", msg=" + msg + ']' ) ; sendWithRetries ( nodeId , msg ) ; } catch ( IgniteCheckedException e ) { if ( e . hasCause ( ClusterTopologyCheckedException . class ) ) { if ( log . isDebugEnabled ( ) ) log . debug ( "Failed to send fragmentizer request to remote node (node left grid): " + nodeId ) ; } else U . error ( log , "Failed to send fragmentizer request to remote node [nodeId=" + nodeId + ", msg=" + msg + ']' , e ) ; nodeIds . remove ( nodeId ) ; } } if ( nodeIds . isEmpty ( ) ) { if ( log . isDebugEnabled ( ) ) log . debug ( "Got empty wait set for fragmentized file: " + fileInfo ) ; fragmentingFiles . remove ( fileInfo . id ( ) , nodeIds ) ; } }
starts file fragmenting . will group file affinity ranges by nodes and send requests to each node . file will be considered processed when each node replied with success ( or error ) or left the grid .
