@Override
  public void rename(String src, String dst, RenamePOptions options)
      throws IOException, AccessControlException {
    UnderFileSystem srcUfs = getUfsInstance(src);
    UnderFileSystem dstUfs = getUfsInstance(dst);
    LOG.debug("Renaming from {} to {}", src, dst);
    // use strong reference comparison as UnderFileSystem does not support equality check
    // except by UFS type
    if (srcUfs != dstUfs) {
      throw new FailedPreconditionException("Cannot rename a file in one UFS to another UFS");
    }

    try {
      // Check if the target file already exists. If yes, return by throwing error.
      boolean overWrite = options.hasOverwrite() ? options.getOverwrite() : false;
      boolean exists = srcUfs.exists(dst);
      if (!overWrite && exists) {
        throw new AlreadyExistsRuntimeException(String.format("File %s already exists but"
            + "no overwrite flag", dst));
      } else if (overWrite) {
        mMetaManager.removeFromMetaStore(dst);
      }
    } catch (IOException e) {
      throw new RuntimeException(e);
    }

    boolean rc;
    try {
      UfsStatus status = srcUfs.getStatus(src);
      if (status.isFile()) {
        rc = srcUfs.renameFile(src, dst);
      } else {
        rc = srcUfs.renameDirectory(src, dst);
      }
      if (rc) {
        mMetaManager.removeFromMetaStore(src);
        mMetaManager.loadFromUfs(dst);
        mMetaManager.invalidateListingCacheOfParent(dst);
      }
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
    LOG.debug("Renaming from {} to {} done: {}", src, dst, rc);
    if (!rc) {
      throw new RuntimeException(String.format("Failed to rename from '%s' to '%s'", src, dst));
    }
  }
@Test
  public void testRename() throws IOException, AccessControlException {
    File srcFolder = mTestFolder.newFolder("root");
    String rootPath = srcFolder.getAbsolutePath();
    mWorker.rename(rootPath, rootPath + "2", RenamePOptions.getDefaultInstance());
    assertFalse(mWorker.exists(rootPath, ExistsPOptions.getDefaultInstance()));
    assertTrue(mWorker.exists(rootPath + "2", ExistsPOptions.getDefaultInstance()));
    String fileContent = "foobar";
    File f = mTestFolder.newFile("root2/f");
    Files.write(f.toPath(), fileContent.getBytes());
    mWorker.rename(f.getAbsolutePath(), f.getAbsolutePath() + "2",
        RenamePOptions.getDefaultInstance());
    assertFalse(mWorker.exists(f.getAbsolutePath(), ExistsPOptions.getDefaultInstance()));
    assertTrue(mWorker.exists(f.getAbsolutePath() + "2", ExistsPOptions.getDefaultInstance()));
  }