@VisibleForTesting
  Properties getTemplateBindings(String userName) {
    Properties k8sProperties = new Properties();

    // k8s template properties
    k8sProperties.put("zeppelin.k8s.interpreter.user", String.valueOf(userName).trim());
    k8sProperties.put("zeppelin.k8s.interpreter.namespace", getInterpreterNamespace());
    k8sProperties.put("zeppelin.k8s.interpreter.pod.name", getPodName());
    k8sProperties.put("zeppelin.k8s.interpreter.serviceAccount", getServiceAccount());
    k8sProperties.put("zeppelin.k8s.interpreter.container.name", interpreterGroupName.toLowerCase());
    k8sProperties.put("zeppelin.k8s.interpreter.container.image", containerImage);
    k8sProperties.put("zeppelin.k8s.interpreter.group.id", getInterpreterGroupId());
    k8sProperties.put("zeppelin.k8s.interpreter.group.name", interpreterGroupName);
    k8sProperties.put("zeppelin.k8s.interpreter.setting.name", getInterpreterSettingName());
    k8sProperties.put("zeppelin.k8s.interpreter.localRepo", getLocalRepoDir());
    k8sProperties.put("zeppelin.k8s.interpreter.rpc.portRange", getInterpreterPortRange());
    k8sProperties.put("zeppelin.k8s.server.rpc.service", intpEventServerHost);
    k8sProperties.put("zeppelin.k8s.server.rpc.portRange", intpEventServerPort);

    String serverNamespace = K8sUtils.getCurrentK8sNamespace();
    String interpreterNamespace = getInterpreterNamespace();
    //Set the owner reference (zeppelin-server pod) for garbage collection when zeppelin server and the zeppelin interpreter is in the same namespace (Kubernetes cannot specify an owner in different namespace).
    if (ownerUID() != null && ownerName() != null && StringUtils.equals(serverNamespace, interpreterNamespace)) {
      k8sProperties.put("zeppelin.k8s.server.uid", ownerUID());
      k8sProperties.put("zeppelin.k8s.server.pod.name", ownerName());
    }

    Map<String, String> k8sEnv = new HashMap<>(getEnv());
    // environment variables
    k8sEnv.put(ENV_SERVICE_DOMAIN, getEnv().getOrDefault(ENV_SERVICE_DOMAIN, System.getenv(ENV_SERVICE_DOMAIN) == null ? "local.zeppelin-project.org" : System.getenv(ENV_SERVICE_DOMAIN)));
    k8sEnv.put(ENV_ZEPPELIN_HOME, getEnv().getOrDefault(ENV_ZEPPELIN_HOME, System.getenv(ENV_ZEPPELIN_HOME)));

    if (isSpark()) {
      int webUiPort = 4040;
      k8sProperties.put("zeppelin.k8s.spark.container.image", sparkImage);

      // There is already initial value following --driver-java-options added in interpreter.sh
      // so we need to pass spark.driver.defaultJavaOptions and spark.driver.extraJavaOptions
      // as SPARK_DRIVER_EXTRAJAVAOPTIONS_CONF env variable to build spark-submit command correctly.
      StringJoiner driverExtraJavaOpts = new StringJoiner(" ");
      if (properties.containsKey(SPARK_DRIVER_DEFAULTJAVAOPTS)) {
        driverExtraJavaOpts.add((String) properties.remove(SPARK_DRIVER_DEFAULTJAVAOPTS));
      }
      if (properties.containsKey(SPARK_DRIVER_EXTRAJAVAOPTS)) {
        driverExtraJavaOpts.add((String) properties.remove(SPARK_DRIVER_EXTRAJAVAOPTS));
      }
      if (driverExtraJavaOpts.length() > 0) {
        k8sEnv.put("SPARK_DRIVER_EXTRAJAVAOPTIONS_CONF", driverExtraJavaOpts.toString());
      }

      if (isSparkOnKubernetes(properties)) {
        addSparkK8sProperties();
        k8sEnv.put("ZEPPELIN_SPARK_CONF", prepareZeppelinSparkConf(userName));
      }
      k8sEnv.put("SPARK_HOME", getEnv().getOrDefault("SPARK_HOME", "/spark"));

      // configure interpreter property "zeppelin.spark.uiWebUrl" if not defined, to enable spark ui through reverse proxy
      String webUrl = (String) properties.get("zeppelin.spark.uiWebUrl");
      if (StringUtils.isBlank(webUrl)) {
        webUrl = "//{{PORT}}-{{SERVICE_NAME}}.{{SERVICE_DOMAIN}}";
      }
      properties.put("zeppelin.spark.uiWebUrl",
          sparkUiWebUrlFromTemplate(
              webUrl,
              webUiPort,
              getPodName(),
              k8sEnv.get(ENV_SERVICE_DOMAIN)
          ));

      // configure interpreter property "zeppelin.k8s.spark.ingress.host" if not defined, to enable spark ui through ingress
      String ingressHost = (String) properties.get("zeppelin.k8s.spark.ingress.host");
      if (StringUtils.isBlank(ingressHost)) {
        ingressHost = "{{PORT}}-{{SERVICE_NAME}}.{{SERVICE_DOMAIN}}";
      }
      properties.put("zeppelin.k8s.spark.ingress.host",
          sparkUiWebUrlFromTemplate(
              ingressHost,
              webUiPort,
              getPodName(),
              k8sEnv.get(ENV_SERVICE_DOMAIN)
          ));

      // Resources of Interpreter Pod
      if (properties.containsKey(SPARK_DRIVER_MEMORY)) {
        String memory;
        if (properties.containsKey(SPARK_DRIVER_MEMORY_OVERHEAD)) {
          memory = K8sUtils.calculateSparkMemory(properties.getProperty(SPARK_DRIVER_MEMORY),
                                                 properties.getProperty(SPARK_DRIVER_MEMORY_OVERHEAD));
        } else {
          memory = K8sUtils.calculateMemoryWithDefaultOverhead(properties.getProperty(SPARK_DRIVER_MEMORY));
        }
        k8sProperties.put("zeppelin.k8s.interpreter.memory", memory);
      }
      if (properties.containsKey(SPARK_DRIVER_CORES)) {
        k8sProperties.put("zeppelin.k8s.interpreter.cores", properties.getProperty(SPARK_DRIVER_CORES));
      }
    }

    k8sProperties.put("zeppelin.k8s.envs", k8sEnv);

    // interpreter properties overrides the values
    k8sProperties.putAll(Maps.fromProperties(properties));
    return k8sProperties;
  }
@Test
  void testGetTemplateBindings() {
    // given
    Properties properties = new Properties();
    properties.put("my.key1", "v1");
    Map<String, String> envs = new HashMap<>();
    envs.put("MY_ENV1", "V1");

    K8sRemoteInterpreterProcess intp = new K8sRemoteInterpreterProcess(
        client,
        "default",
        new File(".skip"),
        "interpreter-container:1.0",
        "shared_process",
        "sh",
        "shell",
        properties,
        envs,
        "zeppelin.server.service",
        12320,
        false,
        "spark-container:1.0",
        10,
        10,
        false,
        false);

    // when
    Properties p = intp.getTemplateBindings(null);

    // then
    assertEquals("default", p.get("zeppelin.k8s.interpreter.namespace"));
    assertEquals(intp.getPodName(), p.get("zeppelin.k8s.interpreter.pod.name"));
    assertEquals("sh", p.get("zeppelin.k8s.interpreter.container.name"));
    assertEquals("interpreter-container:1.0", p.get("zeppelin.k8s.interpreter.container.image"));
    assertEquals("shared_process", p.get("zeppelin.k8s.interpreter.group.id"));
    assertEquals("sh", p.get("zeppelin.k8s.interpreter.group.name"));
    assertEquals("shell", p.get("zeppelin.k8s.interpreter.setting.name"));
    assertTrue(p.containsKey("zeppelin.k8s.interpreter.localRepo"));
    assertEquals("12321:12321" , p.get("zeppelin.k8s.interpreter.rpc.portRange"));
    assertEquals("zeppelin.server.service" , p.get("zeppelin.k8s.server.rpc.service"));
    assertEquals(12320 , p.get("zeppelin.k8s.server.rpc.portRange"));
    assertEquals("null", p.get("zeppelin.k8s.interpreter.user"));
    assertEquals("v1", p.get("my.key1"));
    assertEquals("V1", envs.get("MY_ENV1"));

    envs = (HashMap<String, String>) p.get("zeppelin.k8s.envs");
    assertTrue(envs.containsKey("SERVICE_DOMAIN"));
    assertTrue(envs.containsKey("ZEPPELIN_HOME"));
    intp.close();
  }