@VisibleForTesting
    Map<String, String> generateApplicationMasterEnv(
            final YarnApplicationFileUploader fileUploader,
            final String classPathStr,
            final String localFlinkJarStr,
            final String appIdStr)
            throws IOException {
        final Map<String, String> env = new HashMap<>();
        // set user specified app master environment variables
        env.putAll(
                ConfigurationUtils.getPrefixedKeyValuePairs(
                        ResourceManagerOptions.CONTAINERIZED_MASTER_ENV_PREFIX,
                        this.flinkConfiguration));
        // set Flink app class path
        env.put(ENV_FLINK_CLASSPATH, classPathStr);
        // Set FLINK_LIB_DIR to `lib` folder under working dir in container
        env.put(ENV_FLINK_LIB_DIR, Path.CUR_DIR + "/" + ConfigConstants.DEFAULT_FLINK_LIB_DIR);
        // Set FLINK_OPT_DIR to `opt` folder under working dir in container
        env.put(ENV_FLINK_OPT_DIR, Path.CUR_DIR + "/" + ConfigConstants.DEFAULT_FLINK_OPT_DIR);
        // set Flink on YARN internal configuration values
        env.put(YarnConfigKeys.FLINK_DIST_JAR, localFlinkJarStr);
        env.put(YarnConfigKeys.ENV_APP_ID, appIdStr);
        env.put(YarnConfigKeys.ENV_CLIENT_HOME_DIR, fileUploader.getHomeDir().toString());
        env.put(
                YarnConfigKeys.ENV_CLIENT_SHIP_FILES,
                encodeYarnLocalResourceDescriptorListToString(
                        fileUploader.getEnvShipResourceList()));
        env.put(
                YarnConfigKeys.FLINK_YARN_FILES,
                fileUploader.getApplicationDir().toUri().toString());
        // https://github.com/apache/hadoop/blob/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/YarnApplicationSecurity.md#identity-on-an-insecure-cluster-hadoop_user_name
        env.put(
                YarnConfigKeys.ENV_HADOOP_USER_NAME,
                UserGroupInformation.getCurrentUser().getUserName());
        // set classpath from YARN configuration
        Utils.setupYarnClassPath(this.yarnConfiguration, env);
        return env;
    }
@Test
    public void testGenerateApplicationMasterEnv(@TempDir File flinkHomeDir) throws IOException {
        final String fakeLocalFlinkJar = "./lib/flink_dist.jar";
        final String fakeClassPath = fakeLocalFlinkJar + ":./usrlib/user.jar";
        final ApplicationId appId = ApplicationId.newInstance(0, 0);
        final Map<String, String> masterEnv =
                getTestMasterEnv(
                        new Configuration(), flinkHomeDir, fakeClassPath, fakeLocalFlinkJar, appId);

        assertThat(masterEnv)
                .containsEntry(ConfigConstants.ENV_FLINK_LIB_DIR, "./lib")
                .containsEntry(YarnConfigKeys.ENV_APP_ID, appId.toString())
                .containsEntry(
                        YarnConfigKeys.FLINK_YARN_FILES,
                        YarnApplicationFileUploader.getApplicationDirPath(
                                        new Path(flinkHomeDir.getPath()), appId)
                                .toString())
                .containsEntry(YarnConfigKeys.ENV_FLINK_CLASSPATH, fakeClassPath);
        assertThat(masterEnv.get(YarnConfigKeys.ENV_CLIENT_SHIP_FILES)).isEmpty();
        assertThat(masterEnv)
                .containsEntry(YarnConfigKeys.FLINK_DIST_JAR, fakeLocalFlinkJar)
                .containsEntry(YarnConfigKeys.ENV_CLIENT_HOME_DIR, flinkHomeDir.getPath());
    }