@Override
  protected void rename(
      List<HadoopResourceId> srcResourceIds,
      List<HadoopResourceId> destResourceIds,
      MoveOptions... moveOptions)
      throws IOException {
    if (moveOptions.length > 0) {
      throw new UnsupportedOperationException("Support for move options is not yet implemented.");
    }
    for (int i = 0; i < srcResourceIds.size(); ++i) {

      final Path srcPath = srcResourceIds.get(i).toPath();
      final Path destPath = destResourceIds.get(i).toPath();

      // this enforces src and dest file systems to match
      final org.apache.hadoop.fs.FileSystem fs = srcPath.getFileSystem(configuration);

      // rename in HDFS requires the target directory to exist or silently fails (BEAM-4861)
      mkdirs(destPath);

      boolean success = fs.rename(srcPath, destPath);

      // If the failure was due to the file already existing, delete and retry (BEAM-5036).
      // This should be the exceptional case, so handle here rather than incur the overhead of
      // testing first
      if (!success && fs.exists(srcPath) && fs.exists(destPath)) {
        LOG.debug(LOG_DELETING_EXISTING_FILE, Path.getPathWithoutSchemeAndAuthority(destPath));
        fs.delete(destPath, false); // not recursive
        success = fs.rename(srcPath, destPath);
      }

      if (!success) {
        if (!fs.exists(srcPath)) {
          throw new FileNotFoundException(
              String.format(
                  "Unable to rename resource %s to %s as source not found.", srcPath, destPath));

        } else if (fs.exists(destPath)) {
          throw new FileAlreadyExistsException(
              String.format(
                  "Unable to rename resource %s to %s as destination already exists and couldn't be deleted.",
                  srcPath, destPath));

        } else {
          throw new IOException(
              String.format(
                  "Unable to rename resource %s to %s. No further information provided by underlying filesystem.",
                  srcPath, destPath));
        }
      }
    }
  }
@Test(expected = FileNotFoundException.class)
  public void testRenameRetryScenario() throws Exception {
    testRename();
    // retry the knowing that sources are already moved to destination
    fileSystem.rename(
        ImmutableList.of(testPath("testFileA"), testPath("testFileB")),
        ImmutableList.of(testPath("renameFileA"), testPath("renameFileB")));
  }