@VisibleForTesting
  long copyBytes(CopyListingFileStatus source2, long sourceOffset,
      OutputStream outStream, int bufferSize, Mapper.Context context)
      throws IOException {
    Path source = source2.getPath();
    byte buf[] = new byte[bufferSize];
    ThrottledInputStream inStream = null;
    long totalBytesRead = 0;

    long chunkLength = source2.getChunkLength();
    boolean finished = false;
    try {
      inStream = getInputStream(source, context.getConfiguration());
      long fileLength = source2.getLen();
      int numBytesToRead  = (int) getNumBytesToRead(fileLength, sourceOffset,
              bufferSize);
      seekIfRequired(inStream, sourceOffset);
      int bytesRead = readBytes(inStream, buf, numBytesToRead);
      while (bytesRead > 0) {
        if (chunkLength > 0 &&
            (totalBytesRead + bytesRead) >= chunkLength) {
          bytesRead = (int)(chunkLength - totalBytesRead);
          finished = true;
        }
        totalBytesRead += bytesRead;
        sourceOffset += bytesRead;
        outStream.write(buf, 0, bytesRead);
        updateContextStatus(totalBytesRead, context, source2);
        if (finished) {
          break;
        }
        numBytesToRead  = (int) getNumBytesToRead(fileLength, sourceOffset,
                bufferSize);
        bytesRead = readBytes(inStream, buf, numBytesToRead);
      }
      outStream.close();
      outStream = null;
    } finally {
      IOUtils.cleanupWithLogger(LOG, outStream, inStream);
    }
    return totalBytesRead;
  }
@SuppressWarnings("rawtypes")
  @Test
  public void testFailOnCloseError() throws Exception {
    Mapper.Context context = mock(Mapper.Context.class);
    doReturn(new Configuration()).when(context).getConfiguration();

    Exception expectedEx = new IOException("boom");
    OutputStream out = mock(OutputStream.class);
    doThrow(expectedEx).when(out).close();

    File f = File.createTempFile(this.getClass().getSimpleName(), null);
    f.deleteOnExit();
    CopyListingFileStatus stat = new CopyListingFileStatus(
        new FileStatus(1L, false, 1, 1024, 0, new Path(f.toURI())));
    
    Exception actualEx = null;
    try {
      new RetriableFileCopyCommand("testFailOnCloseError", FileAction.OVERWRITE)
        .copyBytes(stat, 0, out, 512, context);
    } catch (Exception e) {
      actualEx = e;
    }
    assertNotNull("close didn't fail", actualEx);
    assertEquals(expectedEx, actualEx);
  }