FadvisedFileRegion extends DefaultFileRegion { @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) { if(readSize < trans) { trans -= readSize; position += readSize; byteBuffer.flip(); } else { byteBuffer.limit((int)trans); byteBuffer.position(0); position += trans; trans = 0; } while(byteBuffer.hasRemaining()) { target.write(byteBuffer); } byteBuffer.clear(); } return actualCount - trans; } FadvisedFileRegion(RandomAccessFile file, long position, long count,
      boolean manageOsCache, int readaheadLength, ReadaheadPool readaheadPool,
      String identifier, int shuffleBufferSize, 
      boolean shuffleTransferToAllowed); @Override long transferTo(WritableByteChannel target, long position); @Override void releaseExternalResources(); void transferSuccessful();  }
@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"); 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); } 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()); fileRegion = new FadvisedFileRegion( inputFile, position, count, false, 0, null, null, 1024, false); customShuffleTransferCornerCases(fileRegion, target, count); long pos = 0; long size; while((size = fileRegion.customShuffleTransfer(target, pos)) > 0) { pos += size; } Assert.assertEquals(count, (int)pos); Assert.assertEquals(count, targetFile.length()); } finally { if (fileRegion != null) { fileRegion.releaseExternalResources(); } IOUtils.cleanupWithLogger(LOG, target); IOUtils.cleanupWithLogger(LOG, targetFile); IOUtils.cleanupWithLogger(LOG, inputFile); } 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); } inFile.delete(); outFile.delete(); testDir.delete(); absLogDir.delete(); }