ContainerLaunchContext setupApplicationMasterContainer(
            String yarnClusterEntrypoint, boolean hasKrb5, JobManagerProcessSpec processSpec) {
        // ------------------ Prepare Application Master Container  ------------------------------

        // respect custom JVM options in the YAML file
        List<ConfigOption<String>> jvmOptions =
                Arrays.asList(
                        CoreOptions.FLINK_DEFAULT_JVM_OPTIONS,
                        CoreOptions.FLINK_JVM_OPTIONS,
                        CoreOptions.FLINK_DEFAULT_JM_JVM_OPTIONS,
                        CoreOptions.FLINK_JM_JVM_OPTIONS);
        String javaOpts = Utils.generateJvmOptsString(flinkConfiguration, jvmOptions, hasKrb5);

        // Set up the container launch context for the application master
        ContainerLaunchContext amContainer = Records.newRecord(ContainerLaunchContext.class);

        final Map<String, String> startCommandValues = new HashMap<>();
        startCommandValues.put("java", "$JAVA_HOME/bin/java");

        String jvmHeapMem =
                JobManagerProcessUtils.generateJvmParametersStr(processSpec, flinkConfiguration);
        startCommandValues.put("jvmmem", jvmHeapMem);

        startCommandValues.put("jvmopts", javaOpts);
        startCommandValues.put(
                "logging", YarnLogConfigUtil.getLoggingYarnCommand(flinkConfiguration));

        startCommandValues.put("class", yarnClusterEntrypoint);
        startCommandValues.put(
                "redirects",
                "1> "
                        + ApplicationConstants.LOG_DIR_EXPANSION_VAR
                        + "/jobmanager.out "
                        + "2> "
                        + ApplicationConstants.LOG_DIR_EXPANSION_VAR
                        + "/jobmanager.err");
        String dynamicParameterListStr =
                JobManagerProcessUtils.generateDynamicConfigsStr(processSpec);
        startCommandValues.put("args", dynamicParameterListStr);

        final String commandTemplate =
                flinkConfiguration.get(YARN_CONTAINER_START_COMMAND_TEMPLATE);
        final String amCommand = getStartCommand(commandTemplate, startCommandValues);

        amContainer.setCommands(Collections.singletonList(amCommand));

        LOG.debug("Application Master start command: " + amCommand);

        return amContainer;
    }
@Test
    void testSetupApplicationMasterContainer() {
        Configuration cfg = new Configuration();
        YarnClusterDescriptor clusterDescriptor = createYarnClusterDescriptor(cfg);

        final JobManagerProcessSpec jobManagerProcessSpec =
                createDefaultJobManagerProcessSpec(1024);
        final String java = "$JAVA_HOME/bin/java";
        final String jvmmem =
                JobManagerProcessUtils.generateJvmParametersStr(jobManagerProcessSpec, cfg);
        final String dynamicParameters =
                JobManagerProcessUtils.generateDynamicConfigsStr(jobManagerProcessSpec);
        final String defaultJvmOpts = "-DdefaultJvm"; // if set
        final String jvmOpts = "-Djvm"; // if set
        final String defaultJmJvmOpts = "-DdefaultJmJvm"; // if set
        final String jmJvmOpts = "-DjmJvm"; // if set
        final String krb5 = "-Djava.security.krb5.conf=krb5.conf";
        final String logfile =
                "-Dlog.file=\""
                        + ApplicationConstants.LOG_DIR_EXPANSION_VAR
                        + "/jobmanager.log\""; // if set
        final String logback =
                "-Dlogback.configurationFile=file:"
                        + YarnLogConfigUtil.CONFIG_FILE_LOGBACK_NAME; // if set
        final String log4j =
                "-Dlog4j.configuration=file:"
                        + YarnLogConfigUtil.CONFIG_FILE_LOG4J_NAME
                        + " -Dlog4j.configurationFile=file:"
                        + YarnLogConfigUtil.CONFIG_FILE_LOG4J_NAME; // if set
        final String mainClass = clusterDescriptor.getYarnSessionClusterEntrypoint();
        final String redirects =
                "1> "
                        + ApplicationConstants.LOG_DIR_EXPANSION_VAR
                        + "/jobmanager.out "
                        + "2> "
                        + ApplicationConstants.LOG_DIR_EXPANSION_VAR
                        + "/jobmanager.err";

        try {
            // no logging, with/out krb5
            assertThat(
                            clusterDescriptor
                                    .setupApplicationMasterContainer(
                                            mainClass, false, jobManagerProcessSpec)
                                    .getCommands()
                                    .get(0))
                    .isEqualTo(
                            String.join(
                                    " ",
                                    java,
                                    jvmmem,
                                    YarnClusterDescriptor.IGNORE_UNRECOGNIZED_VM_OPTIONS,
                                    mainClass,
                                    dynamicParameters,
                                    redirects));

            assertThat(
                            clusterDescriptor
                                    .setupApplicationMasterContainer(
                                            mainClass, true, jobManagerProcessSpec)
                                    .getCommands()
                                    .get(0))
                    .isEqualTo(
                            String.join(
                                    " ",
                                    java,
                                    jvmmem,
                                    YarnClusterDescriptor.IGNORE_UNRECOGNIZED_VM_OPTIONS,
                                    krb5,
                                    mainClass,
                                    dynamicParameters,
                                    redirects));

            // logback only, with/out krb5
            cfg.set(
                    YarnConfigOptionsInternal.APPLICATION_LOG_CONFIG_FILE,
                    YarnLogConfigUtil.CONFIG_FILE_LOGBACK_NAME);
            assertThat(
                            clusterDescriptor
                                    .setupApplicationMasterContainer(
                                            mainClass, false, jobManagerProcessSpec)
                                    .getCommands()
                                    .get(0))
                    .isEqualTo(
                            String.join(
                                    " ",
                                    java,
                                    jvmmem,
                                    YarnClusterDescriptor.IGNORE_UNRECOGNIZED_VM_OPTIONS,
                                    logfile,
                                    logback,
                                    mainClass,
                                    dynamicParameters,
                                    redirects));

            cfg.set(
                    YarnConfigOptionsInternal.APPLICATION_LOG_CONFIG_FILE,
                    YarnLogConfigUtil.CONFIG_FILE_LOGBACK_NAME);
            assertThat(
                            clusterDescriptor
                                    .setupApplicationMasterContainer(
                                            mainClass, true, jobManagerProcessSpec)
                                    .getCommands()
                                    .get(0))
                    .isEqualTo(
                            String.join(
                                    " ",
                                    java,
                                    jvmmem,
                                    YarnClusterDescriptor.IGNORE_UNRECOGNIZED_VM_OPTIONS,
                                    krb5,
                                    logfile,
                                    logback,
                                    mainClass,
                                    dynamicParameters,
                                    redirects));

            // log4j, with/out krb5
            cfg.set(
                    YarnConfigOptionsInternal.APPLICATION_LOG_CONFIG_FILE,
                    YarnLogConfigUtil.CONFIG_FILE_LOG4J_NAME);
            assertThat(
                            clusterDescriptor
                                    .setupApplicationMasterContainer(
                                            mainClass, false, jobManagerProcessSpec)
                                    .getCommands()
                                    .get(0))
                    .isEqualTo(
                            String.join(
                                    " ",
                                    java,
                                    jvmmem,
                                    YarnClusterDescriptor.IGNORE_UNRECOGNIZED_VM_OPTIONS,
                                    logfile,
                                    log4j,
                                    mainClass,
                                    dynamicParameters,
                                    redirects));

            cfg.set(
                    YarnConfigOptionsInternal.APPLICATION_LOG_CONFIG_FILE,
                    YarnLogConfigUtil.CONFIG_FILE_LOG4J_NAME);
            assertThat(
                            clusterDescriptor
                                    .setupApplicationMasterContainer(
                                            mainClass, true, jobManagerProcessSpec)
                                    .getCommands()
                                    .get(0))
                    .isEqualTo(
                            String.join(
                                    " ",
                                    java,
                                    jvmmem,
                                    YarnClusterDescriptor.IGNORE_UNRECOGNIZED_VM_OPTIONS,
                                    krb5,
                                    logfile,
                                    log4j,
                                    mainClass,
                                    dynamicParameters,
                                    redirects));

            // logback, with/out krb5
            cfg.set(
                    YarnConfigOptionsInternal.APPLICATION_LOG_CONFIG_FILE,
                    YarnLogConfigUtil.CONFIG_FILE_LOGBACK_NAME);
            assertThat(
                            clusterDescriptor
                                    .setupApplicationMasterContainer(
                                            mainClass, false, jobManagerProcessSpec)
                                    .getCommands()
                                    .get(0))
                    .isEqualTo(
                            String.join(
                                    " ",
                                    java,
                                    jvmmem,
                                    YarnClusterDescriptor.IGNORE_UNRECOGNIZED_VM_OPTIONS,
                                    logfile,
                                    logback,
                                    mainClass,
                                    dynamicParameters,
                                    redirects));

            cfg.set(
                    YarnConfigOptionsInternal.APPLICATION_LOG_CONFIG_FILE,
                    YarnLogConfigUtil.CONFIG_FILE_LOGBACK_NAME);
            assertThat(
                            clusterDescriptor
                                    .setupApplicationMasterContainer(
                                            mainClass, true, jobManagerProcessSpec)
                                    .getCommands()
                                    .get(0))
                    .isEqualTo(
                            String.join(
                                    " ",
                                    java,
                                    jvmmem,
                                    YarnClusterDescriptor.IGNORE_UNRECOGNIZED_VM_OPTIONS,
                                    krb5,
                                    logfile,
                                    logback,
                                    mainClass,
                                    dynamicParameters,
                                    redirects));

            // logback, with/out krb5, different JVM opts
            // IMPORTANT: Be aware that we are using side effects here to modify the created
            // YarnClusterDescriptor,
            // because we have a reference to the ClusterDescriptor's configuration which we modify
            // continuously
            cfg.set(CoreOptions.FLINK_DEFAULT_JVM_OPTIONS, defaultJvmOpts);
            cfg.set(CoreOptions.FLINK_JVM_OPTIONS, jvmOpts);
            cfg.set(
                    YarnConfigOptionsInternal.APPLICATION_LOG_CONFIG_FILE,
                    YarnLogConfigUtil.CONFIG_FILE_LOGBACK_NAME);
            assertThat(
                            clusterDescriptor
                                    .setupApplicationMasterContainer(
                                            mainClass, false, jobManagerProcessSpec)
                                    .getCommands()
                                    .get(0))
                    .isEqualTo(
                            String.join(
                                    " ",
                                    java,
                                    jvmmem,
                                    defaultJvmOpts,
                                    jvmOpts,
                                    YarnClusterDescriptor.IGNORE_UNRECOGNIZED_VM_OPTIONS,
                                    logfile,
                                    logback,
                                    mainClass,
                                    dynamicParameters,
                                    redirects));

            cfg.set(
                    YarnConfigOptionsInternal.APPLICATION_LOG_CONFIG_FILE,
                    YarnLogConfigUtil.CONFIG_FILE_LOGBACK_NAME);
            assertThat(
                            clusterDescriptor
                                    .setupApplicationMasterContainer(
                                            mainClass, true, jobManagerProcessSpec)
                                    .getCommands()
                                    .get(0))
                    .isEqualTo(
                            String.join(
                                    " ",
                                    java,
                                    jvmmem,
                                    defaultJvmOpts,
                                    jvmOpts,
                                    YarnClusterDescriptor.IGNORE_UNRECOGNIZED_VM_OPTIONS,
                                    krb5,
                                    logfile,
                                    logback,
                                    mainClass,
                                    dynamicParameters,
                                    redirects));

            // log4j, with/out krb5, different JVM opts
            // IMPORTANT: Be aware that we are using side effects here to modify the created
            // YarnClusterDescriptor
            cfg.set(CoreOptions.FLINK_DEFAULT_JM_JVM_OPTIONS, defaultJmJvmOpts);
            cfg.set(CoreOptions.FLINK_JM_JVM_OPTIONS, jmJvmOpts);
            cfg.set(
                    YarnConfigOptionsInternal.APPLICATION_LOG_CONFIG_FILE,
                    YarnLogConfigUtil.CONFIG_FILE_LOG4J_NAME);
            assertThat(
                            clusterDescriptor
                                    .setupApplicationMasterContainer(
                                            mainClass, false, jobManagerProcessSpec)
                                    .getCommands()
                                    .get(0))
                    .isEqualTo(
                            String.join(
                                    " ",
                                    java,
                                    jvmmem,
                                    defaultJvmOpts,
                                    jvmOpts,
                                    defaultJmJvmOpts,
                                    jmJvmOpts,
                                    YarnClusterDescriptor.IGNORE_UNRECOGNIZED_VM_OPTIONS,
                                    logfile,
                                    log4j,
                                    mainClass,
                                    dynamicParameters,
                                    redirects));

            cfg.set(
                    YarnConfigOptionsInternal.APPLICATION_LOG_CONFIG_FILE,
                    YarnLogConfigUtil.CONFIG_FILE_LOG4J_NAME);
            assertThat(
                            clusterDescriptor
                                    .setupApplicationMasterContainer(
                                            mainClass, true, jobManagerProcessSpec)
                                    .getCommands()
                                    .get(0))
                    .isEqualTo(
                            String.join(
                                    " ",
                                    java,
                                    jvmmem,
                                    defaultJvmOpts,
                                    jvmOpts,
                                    defaultJmJvmOpts,
                                    jmJvmOpts,
                                    YarnClusterDescriptor.IGNORE_UNRECOGNIZED_VM_OPTIONS,
                                    krb5,
                                    logfile,
                                    log4j,
                                    mainClass,
                                    dynamicParameters,
                                    redirects));

            // now try some configurations with different yarn.container-start-command-template
            // IMPORTANT: Be aware that we are using side effects here to modify the created
            // YarnClusterDescriptor
            cfg.set(
                    YarnConfigOptionsInternal.APPLICATION_LOG_CONFIG_FILE,
                    YarnLogConfigUtil.CONFIG_FILE_LOGBACK_NAME);
            cfg.set(
                    YARN_CONTAINER_START_COMMAND_TEMPLATE,
                    "%java% 1 %jvmmem% 2 %jvmopts% 3 %logging% 4 %class% 5 %args% 6 %redirects%");
            assertThat(
                            clusterDescriptor
                                    .setupApplicationMasterContainer(
                                            mainClass, true, jobManagerProcessSpec)
                                    .getCommands()
                                    .get(0))
                    .isEqualTo(
                            String.join(
                                    " ",
                                    java,
                                    "1",
                                    jvmmem,
                                    "2",
                                    defaultJvmOpts,
                                    jvmOpts,
                                    defaultJmJvmOpts,
                                    jmJvmOpts,
                                    YarnClusterDescriptor.IGNORE_UNRECOGNIZED_VM_OPTIONS,
                                    krb5,
                                    "3",
                                    logfile,
                                    logback,
                                    "4",
                                    mainClass,
                                    "5",
                                    dynamicParameters,
                                    "6",
                                    redirects));

            cfg.set(
                    YarnConfigOptionsInternal.APPLICATION_LOG_CONFIG_FILE,
                    YarnLogConfigUtil.CONFIG_FILE_LOGBACK_NAME);
            cfg.set(
                    YARN_CONTAINER_START_COMMAND_TEMPLATE,
                    "%java% %logging% %jvmopts% %jvmmem% %class% %args% %redirects%");
            // IMPORTANT: Be aware that we are using side effects here to modify the created
            // YarnClusterDescriptor
            assertThat(
                            clusterDescriptor
                                    .setupApplicationMasterContainer(
                                            mainClass, true, jobManagerProcessSpec)
                                    .getCommands()
                                    .get(0))
                    .isEqualTo(
                            String.join(
                                    " ",
                                    java,
                                    logfile,
                                    logback,
                                    defaultJvmOpts,
                                    jvmOpts,
                                    defaultJmJvmOpts,
                                    jmJvmOpts,
                                    YarnClusterDescriptor.IGNORE_UNRECOGNIZED_VM_OPTIONS,
                                    krb5,
                                    jvmmem,
                                    mainClass,
                                    dynamicParameters,
                                    redirects));
        } finally {
            clusterDescriptor.close();
        }
    }