public static Optional<Path> getQualifiedRemoteProvidedUsrLib(
            org.apache.flink.configuration.Configuration configuration,
            YarnConfiguration yarnConfiguration)
            throws IOException, IllegalArgumentException {
        String usrlib = configuration.get(YarnConfigOptions.PROVIDED_USRLIB_DIR);
        if (usrlib == null) {
            return Optional.empty();
        }
        final Path qualifiedUsrLibPath =
                FileSystem.get(yarnConfiguration).makeQualified(new Path(usrlib));
        checkArgument(
                isRemotePath(qualifiedUsrLibPath.toString()),
                "The \"%s\" must point to a remote dir "
                        + "which is accessible from all worker nodes.",
                YarnConfigOptions.PROVIDED_USRLIB_DIR.key());
        checkArgument(
                isUsrLibDirectory(FileSystem.get(yarnConfiguration), qualifiedUsrLibPath),
                "The \"%s\" should be named with \"%s\".",
                YarnConfigOptions.PROVIDED_USRLIB_DIR.key(),
                ConfigConstants.DEFAULT_FLINK_USR_LIB_DIR);
        return Optional.of(qualifiedUsrLibPath);
    }
@Test
    void testInvalidRemoteUsrLib(@TempDir Path tempDir) throws IOException {
        final String sharedLibPath = "hdfs:///flink/badlib";

        final org.apache.hadoop.conf.Configuration hdConf =
                new org.apache.hadoop.conf.Configuration();
        hdConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, tempDir.toAbsolutePath().toString());
        try (final MiniDFSCluster hdfsCluster = new MiniDFSCluster.Builder(hdConf).build()) {
            final org.apache.hadoop.fs.Path hdfsRootPath =
                    new org.apache.hadoop.fs.Path(hdfsCluster.getURI());
            hdfsCluster.getFileSystem().mkdirs(new org.apache.hadoop.fs.Path(sharedLibPath));

            final Configuration flinkConfig = new Configuration();
            flinkConfig.set(YarnConfigOptions.PROVIDED_USRLIB_DIR, sharedLibPath);
            final YarnConfiguration yarnConfig = new YarnConfiguration();
            yarnConfig.set(
                    CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY, hdfsRootPath.toString());
            assertThatThrownBy(
                            () -> Utils.getQualifiedRemoteProvidedUsrLib(flinkConfig, yarnConfig))
                    .isInstanceOf(IllegalArgumentException.class)
                    .hasMessage(
                            "The \"%s\" should be named with \"%s\".",
                            YarnConfigOptions.PROVIDED_USRLIB_DIR.key(),
                            ConfigConstants.DEFAULT_FLINK_USR_LIB_DIR);
        }
    }