@Override
  protected void copy(List<HadoopResourceId> srcResourceIds, List<HadoopResourceId> destResourceIds)
      throws IOException {
    for (int i = 0; i < srcResourceIds.size(); ++i) {
      // this enforces src and dest file systems to match
      final org.apache.hadoop.fs.FileSystem fs =
          srcResourceIds.get(i).toPath().getFileSystem(configuration);
      // Unfortunately HDFS FileSystems don't support a native copy operation so we are forced
      // to use the inefficient implementation found in FileUtil which copies all the bytes through
      // the local machine.
      //
      // HDFS FileSystem does define a concat method but could only find the DFSFileSystem
      // implementing it. The DFSFileSystem implemented concat by deleting the srcs after which
      // is not what we want. Also, all the other FileSystem implementations I saw threw
      // UnsupportedOperationException within concat.
      final boolean success =
          FileUtil.copy(
              fs,
              srcResourceIds.get(i).toPath(),
              fs,
              destResourceIds.get(i).toPath(),
              false,
              true,
              fs.getConf());
      if (!success) {
        // Defensive coding as this should not happen in practice
        throw new IOException(
            String.format(
                "Unable to copy resource %s to %s. No further information provided by underlying filesystem.",
                srcResourceIds.get(i).toPath(), destResourceIds.get(i).toPath()));
      }
    }
  }
@Test
  public void testCopy() throws Exception {
    create("testFileA", "testDataA".getBytes(StandardCharsets.UTF_8));
    create("testFileB", "testDataB".getBytes(StandardCharsets.UTF_8));
    fileSystem.copy(
        ImmutableList.of(testPath("testFileA"), testPath("testFileB")),
        ImmutableList.of(testPath("copyTestFileA"), testPath("copyTestFileB")));
    assertArrayEquals("testDataA".getBytes(StandardCharsets.UTF_8), read("testFileA", 0));
    assertArrayEquals("testDataB".getBytes(StandardCharsets.UTF_8), read("testFileB", 0));
    assertArrayEquals("testDataA".getBytes(StandardCharsets.UTF_8), read("copyTestFileA", 0));
    assertArrayEquals("testDataB".getBytes(StandardCharsets.UTF_8), read("copyTestFileB", 0));
  }