public void asyncRenameFiles(
            List<CompletableFuture<?>> renameFileFutures,
            AtomicBoolean cancelled,
            Path writePath,
            Path targetPath,
            List<String> fileNames) {
        FileSystem fileSystem;
        try {
            fileSystem = FileSystem.get(writePath.toUri(), conf);
        } catch (Exception e) {
            Log.error("Failed to get fileSystem", e);
            throw new StarRocksConnectorException("Failed to move data files to target location. " +
                    "Failed to get file system on path %s. msg: %s", writePath, e.getMessage());
        }

        for (String fileName : fileNames) {
            Path source = new Path(writePath, fileName);
            Path target = new Path(targetPath, fileName);
            renameFileFutures.add(CompletableFuture.runAsync(() -> {
                if (cancelled.get()) {
                    return;
                }
                try {
                    if (fileSystem.exists(target)) {
                        throw new StarRocksConnectorException("Failed to move data files from %s to target location %s. msg:" +
                                " target location already exists", source, target);
                    }

                    if (!fileSystem.rename(source, target)) {
                        throw new StarRocksConnectorException("Failed to move data files from %s to target location %s. msg:" +
                                " rename operation failed", source, target);
                    }
                } catch (IOException e) {
                    LOG.error("Failed to rename data files", e);
                    throw new StarRocksConnectorException("Failed to move data files from %s to final location %s. msg: %s",
                            source, target, e.getMessage());
                }
            }, updateRemoteFilesExecutor));
        }
    }
@Test
    public void asyncRenameFilesTest() {
        HiveRemoteFileIO hiveRemoteFileIO = new HiveRemoteFileIO(new Configuration());
        FileSystem fs = new MockedRemoteFileSystem(HDFS_HIVE_TABLE);
        hiveRemoteFileIO.setFileSystem(fs);
        FeConstants.runningUnitTest = true;
        ExecutorService executorToRefresh = Executors.newFixedThreadPool(5);
        ExecutorService executorToLoad = Executors.newFixedThreadPool(5);

        CachingRemoteFileIO cachingFileIO = new CachingRemoteFileIO(hiveRemoteFileIO, executorToRefresh, 10, 10, 10);
        RemoteFileOperations ops = new RemoteFileOperations(cachingFileIO, executorToLoad, executorToLoad,
                false, true, new Configuration());

        List<CompletableFuture<?>> futures = new ArrayList<>();
        Path writePath = new Path("hdfs://hadoop01:9000/tmp/starrocks/queryid");
        Path targetPath = new Path("hdfs://hadoop01:9000/user/hive/warehouse/test.db/t1");
        List<String> fileNames = Lists.newArrayList("file1");
        ExceptionChecker.expectThrowsWithMsg(
                StarRocksConnectorException.class,
                "Failed to move data files to target location." +
                        " Failed to get file system on path hdfs://hadoop01:9000/tmp/starrocks/queryid",
                () -> ops.asyncRenameFiles(futures, new AtomicBoolean(true), writePath, targetPath, fileNames));

        RemoteFileOperations ops1 = new RemoteFileOperations(cachingFileIO, executorToLoad, Executors.newSingleThreadExecutor(),
                false, true, new Configuration());

        FileSystem mockedFs = new MockedRemoteFileSystem(HDFS_HIVE_TABLE) {
            @Override
            public boolean exists(Path path) {
                return true;
            }
        };

        new MockUp<FileSystem>() {
            @Mock
            public FileSystem get(URI uri, Configuration conf) throws IOException {
                return mockedFs;
            }
        };

        ExceptionChecker.expectThrowsWithMsg(
                StarRocksConnectorException.class,
                "Failed to move data files from hdfs://hadoop01:9000/tmp/starrocks/queryid/file1 to" +
                        " target location hdfs://hadoop01:9000/user/hive/warehouse/test.db/t1/file1." +
                        " msg: target location already exists",
                () -> {
                    ops1.asyncRenameFiles(futures, new AtomicBoolean(false), writePath, targetPath, fileNames);
                    getFutureValue(futures.get(0), StarRocksConnectorException.class);
                });

        new MockUp<FileSystem>() {
            @Mock
            public FileSystem get(URI uri, Configuration conf) throws IOException {
                return fs;
            }
        };

        ExceptionChecker.expectThrowsWithMsg(
                StarRocksConnectorException.class,
                "Failed to move data files from hdfs://hadoop01:9000/tmp/starrocks/queryid/file1 to" +
                        " target location hdfs://hadoop01:9000/user/hive/warehouse/test.db/t1/file1." +
                        " msg: rename operation failed",
                () -> {
                    futures.clear();
                    ops.asyncRenameFiles(futures, new AtomicBoolean(false), writePath, targetPath, fileNames);
                    getFutureValue(futures.get(0), StarRocksConnectorException.class);
                });
    }