@VisibleForTesting
  long customShuffleTransfer(WritableByteChannel target, long position)
          throws IOException {
    long actualCount = this.count - position;
    if (actualCount < 0 || position < 0) {
      throw new IllegalArgumentException(
              "position out of range: " + position +
                      " (expected: 0 - " + (this.count - 1) + ')');
    }
    if (actualCount == 0) {
      return 0L;
    }

    long trans = actualCount;
    int readSize;
    ByteBuffer byteBuffer = ByteBuffer.allocate(
            Math.min(
                    this.shuffleBufferSize,
                    trans > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) trans));

    while(trans > 0L &&
            (readSize = fileChannel.read(byteBuffer, this.position+position)) > 0) {
      //adjust counters and buffer limit
      if(readSize < trans) {
        trans -= readSize;
        position += readSize;
        byteBuffer.flip();
      } else {
        //We can read more than we need if the actualCount is not multiple
        //of the byteBuffer size and file is big enough. In that case we cannot
        //use flip method but we need to set buffer limit manually to trans.
        byteBuffer.limit((int)trans);
        byteBuffer.position(0);
        position += trans;
        trans = 0;
      }

      //write data to the target
      while(byteBuffer.hasRemaining()) {
        target.write(byteBuffer);
      }

      byteBuffer.clear();
    }

    return actualCount - trans;
  }
@Test(timeout = 100000)
  public void testCustomShuffleTransfer() throws IOException {
    File absLogDir = new File("target", 
        TestFadvisedFileRegion.class.getSimpleName() + 
        "LocDir").getAbsoluteFile();
    
    String testDirPath =
        StringUtils.join(Path.SEPARATOR,
            new String[] { absLogDir.getAbsolutePath(),
                "testCustomShuffleTransfer"});
    File testDir = new File(testDirPath);
    testDir.mkdirs();
    
    System.out.println(testDir.getAbsolutePath());
    
    File inFile = new File(testDir, "fileIn.out");
    File outFile = new File(testDir, "fileOut.out");
    
    
    //Initialize input file
    byte [] initBuff = new byte[FILE_SIZE];
    Random rand = new Random();
    rand.nextBytes(initBuff);
    
    FileOutputStream out = new FileOutputStream(inFile);
    try{
      out.write(initBuff);  
    } finally {
      IOUtils.cleanupWithLogger(LOG, out);
    }
    
    
    //define position and count to read from a file region.
    int position = 2*1024*1024;
    int count = 4*1024*1024 - 1;
    
    RandomAccessFile inputFile = null;
    RandomAccessFile targetFile = null;
    WritableByteChannel target = null;
    FadvisedFileRegion fileRegion = null;
    
    try {
      inputFile = new RandomAccessFile(inFile.getAbsolutePath(), "r");
      targetFile = new RandomAccessFile(outFile.getAbsolutePath(), "rw");
      target = targetFile.getChannel();
      
      Assert.assertEquals(FILE_SIZE, inputFile.length());
      
      //create FadvisedFileRegion
      fileRegion = new FadvisedFileRegion(
          inputFile, position, count, false, 0, null, null, 1024, false);
      
      //test corner cases
      customShuffleTransferCornerCases(fileRegion, target, count);
            
      long pos = 0;
      long size;
      while((size = fileRegion.customShuffleTransfer(target, pos)) > 0) {
        pos += size; 
      }
    
      //assert size
      Assert.assertEquals(count, (int)pos);
      Assert.assertEquals(count, targetFile.length());
    } finally {
      if (fileRegion != null) {
        fileRegion.deallocate();
      }
      IOUtils.cleanupWithLogger(LOG, target);
      IOUtils.cleanupWithLogger(LOG, targetFile);
      IOUtils.cleanupWithLogger(LOG, inputFile);
    }
    
    //Read the target file and verify that copy is done correctly
    byte [] buff = new byte[FILE_SIZE];
    FileInputStream in = new FileInputStream(outFile);
    try {
      int total = in.read(buff, 0, count);
    
      Assert.assertEquals(count, total);
    
      for(int i = 0; i < count; i++) {
        Assert.assertEquals(initBuff[position+i], buff[i]);
      }
    } finally {
      IOUtils.cleanupWithLogger(LOG, in);
    }
    
    //delete files and folders
    inFile.delete();
    outFile.delete();
    testDir.delete();
    absLogDir.delete();
  }