@Override
  public ListenableFuture<List<RouteFailure>> move(List<Route> routes, UfsReadOptions readOptions,
                                                   WriteOptions writeOptions) {
    List<ListenableFuture<Void>> futures = new ArrayList<>();
    List<RouteFailure> errors = Collections.synchronizedList(new ArrayList<>());
    for (Route route : routes) {
      UnderFileSystem srcUfs = getUnderFileSystem(route.getSrc());
      UnderFileSystem dstUfs = getUnderFileSystem(route.getDst());
      String srcRoot = new AlluxioURI(route.getSrc()).getRootPath();
      String dstRoot = new AlluxioURI(route.getDst()).getRootPath();
      try (FileSystem srcFs = new UfsBaseFileSystem(mFsContext, new UfsFileSystemOptions(srcRoot),
          new UfsManager.UfsClient(() -> srcUfs, new AlluxioURI(srcRoot)));
           FileSystem dstFs = new UfsBaseFileSystem(mFsContext, new UfsFileSystemOptions(dstRoot),
               new UfsManager.UfsClient(() -> dstUfs, new AlluxioURI(dstRoot)))) {
        ListenableFuture<Void> future = Futures.submit(() -> {
          Boolean deleteFailure = false;
          try {
            if (readOptions.hasUser()) {
              AuthenticatedClientUser.set(readOptions.getUser());
            }
            checkMovePermission(route.getSrc(), route.getDst());
            if (!ValidateHandler.validate(route, writeOptions, srcFs, dstFs, true)) {
              throw new FailedPreconditionRuntimeException("File " + route.getDst()
                  + " is already in UFS");
            }
            CopyHandler.copy(route, writeOptions, srcFs, dstFs);
            try {
              DeleteHandler.delete(new AlluxioURI(route.getSrc()), srcFs);
            } catch (Exception e) {
              deleteFailure = true;
              throw e;
            }
          } catch (Throwable t) {
            LOG.error("Failed to move {} to {}", route.getSrc(), route.getDst(), t);
            boolean permissionCheckSucceeded = !(t instanceof AccessControlException);
            AlluxioRuntimeException e = AlluxioRuntimeException.from(t);
            RouteFailure.Builder builder =
                RouteFailure.newBuilder().setRoute(route).setCode(e.getStatus().getCode().value())
                    .setRetryable(e.isRetryable() && permissionCheckSucceeded);
            if (e.getMessage() != null) {
              builder.setMessage(e.getMessage());
            }
            if (deleteFailure) {
              builder.setRetryable(false);
            }
            errors.add(builder.build());
          }
        }, GrpcExecutors.WRITER_EXECUTOR);
        futures.add(future);
      } catch (IOException e) {
        // ignore close error
      } catch (RejectedExecutionException e) {
        LOG.warn("BlockDataWriterExecutor overloaded.");
        AlluxioRuntimeException t = AlluxioRuntimeException.from(e);
        RouteFailure.Builder builder =
            RouteFailure.newBuilder().setRoute(route).setCode(t.getStatus().getCode().value())
                .setRetryable(true);
        errors.add(builder.build());
      }
    }
    return Futures.whenAllComplete(futures).call(() -> errors, GrpcExecutors.WRITER_EXECUTOR);
  }
@Test
  public void testSingleFileMoveWithFileAlreadyExist() throws IOException, ExecutionException,
      InterruptedException {
    File srcRoot = mTestFolder.newFolder("src");
    File dstRoot = mTestFolder.newFolder("dst");
    // create test file under mSrcFolder
    File a = new File(srcRoot, "a");
    a.createNewFile();
    File b = new File(dstRoot, "b");
    b.createNewFile();
    int length = 10;
    byte[] buffer = BufferUtils.getIncreasingByteArray(length);
    BufferUtils.writeBufferToFile(a.getAbsolutePath(), buffer);
    Route route =
        Route.newBuilder().setDst(b.getAbsolutePath()).setSrc(a.getAbsolutePath())
            .setLength(length).build();
    WriteOptions writeOptions =
        WriteOptions.newBuilder().setOverwrite(false).setCheckContent(true).build();
    UfsReadOptions read =
        UfsReadOptions.newBuilder().setUser("test").setTag("1").setPositionShort(false).build();
    ListenableFuture<List<RouteFailure>> move =
        mWorker.move(Collections.singletonList(route), read, writeOptions);
    List<RouteFailure> failures = move.get();
    assertEquals(1, failures.size());
    assertFalse(failures.get(0).hasIsSkip());
  }