@VisibleForTesting
    Path getStagingDir(FileSystem defaultFileSystem) throws IOException {
        final String configuredStagingDir =
                flinkConfiguration.get(YarnConfigOptions.STAGING_DIRECTORY);
        if (configuredStagingDir == null) {
            return defaultFileSystem.getHomeDirectory();
        }
        FileSystem stagingDirFs =
                new Path(configuredStagingDir).getFileSystem(defaultFileSystem.getConf());
        return stagingDirFs.makeQualified(new Path(configuredStagingDir));
    }
@Test
    void testGetStagingDirWithSpecifyingStagingDir() throws IOException {
        final Configuration flinkConfig = new Configuration();
        flinkConfig.set(YarnConfigOptions.STAGING_DIRECTORY, "file:///tmp/path1");
        try (final YarnClusterDescriptor yarnClusterDescriptor =
                createYarnClusterDescriptor(flinkConfig)) {
            YarnConfiguration yarnConfig = new YarnConfiguration();
            yarnConfig.set("fs.defaultFS", "viewfs://hadoop-ns01");
            yarnConfig.set("fs.viewfs.mounttable.hadoop-ns01.link./tmp", "file://tmp");
            FileSystem defaultFileSystem = FileSystem.get(yarnConfig);

            Path stagingDir = yarnClusterDescriptor.getStagingDir(defaultFileSystem);

            assertThat(defaultFileSystem.getScheme()).isEqualTo("viewfs");
            assertThat(stagingDir.getFileSystem(yarnConfig).getScheme()).isEqualTo("file");
        }
    }