private String getEnv(String envName, InterpreterLaunchContext context) {
    String env = context.getProperties().getProperty(envName);
    if (StringUtils.isBlank(env)) {
      env = System.getenv(envName);
    }
    if (StringUtils.isBlank(env)) {
      LOGGER.warn("environment variable: {} is empty", envName);
    }

    return env;
  }
@Test
  void testYarnClusterMode_3() throws IOException {
    SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
    Properties properties = new Properties();
    properties.setProperty("SPARK_HOME", sparkHome);
    properties.setProperty("property_1", "value_1");
    properties.setProperty("spark.master", "yarn");
    properties.setProperty("spark.submit.deployMode", "cluster");
    properties.setProperty("spark.files", "{}");
    properties.setProperty("spark.jars", "jar_1");

    InterpreterOption option = new InterpreterOption();
    option.setUserImpersonate(true);
    InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "user1", "intpGroupId", "groupId", "spark", "spark", 0, "host");
    Path localRepoPath = Paths.get(zConf.getInterpreterLocalRepoPath(), context.getInterpreterSettingId());
    FileUtils.deleteDirectory(localRepoPath.toFile());
    Files.createDirectories(localRepoPath);

    InterpreterClient client = launcher.launch(context);
    assertTrue(client instanceof ExecRemoteInterpreterProcess);
    try (ExecRemoteInterpreterProcess interpreterProcess = (ExecRemoteInterpreterProcess) client) {
      assertEquals("spark", interpreterProcess.getInterpreterSettingName());
      assertTrue(interpreterProcess.getInterpreterDir().endsWith("/interpreter/spark"));
      assertTrue(interpreterProcess.getLocalRepoDir().endsWith("/local-repo/groupId"));
      assertEquals(zConf.getInterpreterRemoteRunnerPath(), interpreterProcess.getInterpreterRunner());
      assertTrue(interpreterProcess.getEnv().size() >= 3);
      assertEquals(sparkHome, interpreterProcess.getEnv().get("SPARK_HOME"));
      assertEquals("true", interpreterProcess.getEnv().get("ZEPPELIN_SPARK_YARN_CLUSTER"));

      String sparkJars = "jar_1," +
        zeppelinHome + "/interpreter/spark/scala-2.12/spark-scala-2.12-" + Util.getVersion()
        + ".jar," +
              zeppelinHome + "/interpreter/zeppelin-interpreter-shaded-" + Util.getVersion() + ".jar";
      String sparkrZip = sparkHome + "/R/lib/sparkr.zip#sparkr";
      // escape special characters
      String sparkFiles = "{}," + zeppelinHome + "/conf/log4j_yarn_cluster.properties";
      String expected = "--proxy-user|user1" +
                      "|--conf|spark.yarn.dist.archives=" + sparkrZip +
                      "|--conf|spark.yarn.isPython=true" +
                      "|--conf|spark.app.name=intpGroupId" +
                      "|--conf|spark.yarn.maxAppAttempts=1" +
                      "|--conf|spark.master=yarn" +
                      "|--conf|spark.files=" + sparkFiles +
                      "|--conf|spark.jars=" + sparkJars +
                      "|--conf|spark.submit.deployMode=cluster" +
                      "|--conf|spark.yarn.submit.waitAppCompletion=false";
      assertTrue(CollectionUtils.isEqualCollection(Arrays.asList(expected.split("\\|")),
        Arrays.asList(interpreterProcess.getEnv().get("ZEPPELIN_SPARK_CONF").split("\\|"))));
      assertTrue(interpreterProcess.getEnv().get("ZEPPELIN_SPARK_CONF").startsWith("--proxy-user|user1"));
    }
    FileUtils.deleteDirectory(localRepoPath.toFile());
  }