public static List<Path> getQualifiedRemoteProvidedLibDirs(
            org.apache.flink.configuration.Configuration configuration,
            YarnConfiguration yarnConfiguration)
            throws IOException {

        return getRemoteSharedLibPaths(
                configuration,
                pathStr -> {
                    final Path path = new Path(pathStr);
                    return path.getFileSystem(yarnConfiguration).makeQualified(path);
                });
    }
@Test
    void testSharedLibWithNonQualifiedPath() throws Exception {
        final String sharedLibPath = "/flink/sharedLib";
        final String nonQualifiedPath = "hdfs://" + sharedLibPath;
        final String defaultFs = "hdfs://localhost:9000";
        final String qualifiedPath = defaultFs + sharedLibPath;

        final Configuration flinkConfig = new Configuration();
        flinkConfig.set(
                YarnConfigOptions.PROVIDED_LIB_DIRS, Collections.singletonList(nonQualifiedPath));
        final YarnConfiguration yarnConfig = new YarnConfiguration();
        yarnConfig.set(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY, defaultFs);

        final List<org.apache.hadoop.fs.Path> sharedLibs =
                Utils.getQualifiedRemoteProvidedLibDirs(flinkConfig, yarnConfig);
        assertThat(sharedLibs).hasSize(1);
        assertThat(sharedLibs.get(0).toUri()).hasToString(qualifiedPath);
    }