http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/InterpreterRestApi.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/InterpreterRestApi.java
 
b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/InterpreterRestApi.java
index c1dba5c..e2a10e6 100644
--- 
a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/InterpreterRestApi.java
+++ 
b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/InterpreterRestApi.java
@@ -123,7 +123,7 @@ public class InterpreterRestApi {
               request.getOption(), request.getProperties());
       logger.info("new setting created with {}", interpreterSetting.getId());
       return new JsonResponse<>(Status.OK, "", interpreterSetting).build();
-    } catch (InterpreterException | IOException e) {
+    } catch (IOException e) {
       logger.error("Exception in InterpreterRestApi while creating ", e);
       return new JsonResponse<>(Status.NOT_FOUND, e.getMessage(), 
ExceptionUtils.getStackTrace(e))
           .build();

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinSparkClusterTest.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinSparkClusterTest.java
 
b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinSparkClusterTest.java
index 3e46449..2fa584b 100644
--- 
a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinSparkClusterTest.java
+++ 
b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinSparkClusterTest.java
@@ -27,6 +27,7 @@ import java.util.List;
 import java.util.Map;
 
 import org.apache.commons.io.FileUtils;
+import org.apache.zeppelin.interpreter.InterpreterException;
 import org.apache.zeppelin.interpreter.InterpreterResult;
 import org.apache.zeppelin.interpreter.InterpreterSetting;
 import org.apache.zeppelin.notebook.Note;
@@ -171,7 +172,7 @@ public class ZeppelinSparkClusterTest extends 
AbstractTestRestApi {
     }
 
     @Test
-    public void sparkRTest() throws IOException {
+    public void sparkRTest() throws IOException, InterpreterException {
       // create new note
       Note note = ZeppelinServer.notebook.createNote(anonymous);
       int sparkVersion = getSparkVersionNumber(note);
@@ -426,7 +427,7 @@ public class ZeppelinSparkClusterTest extends 
AbstractTestRestApi {
     }
 
     @Test
-    public void pySparkDepLoaderTest() throws IOException {
+    public void pySparkDepLoaderTest() throws IOException, 
InterpreterException {
         // create new note
         Note note = ZeppelinServer.notebook.createNote(anonymous);
         int sparkVersionNumber = getSparkVersionNumber(note);

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/pom.xml
----------------------------------------------------------------------
diff --git a/zeppelin-zengine/pom.xml b/zeppelin-zengine/pom.xml
index c67df6b..d1a2270 100644
--- a/zeppelin-zengine/pom.xml
+++ b/zeppelin-zengine/pom.xml
@@ -71,11 +71,6 @@
     </dependency>
 
     <dependency>
-      <groupId>commons-configuration</groupId>
-      <artifactId>commons-configuration</artifactId>
-    </dependency>
-
-    <dependency>
       <groupId>commons-io</groupId>
       <artifactId>commons-io</artifactId>
     </dependency>

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
deleted file mode 100644
index 3a82bc5..0000000
--- 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
+++ /dev/null
@@ -1,847 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zeppelin.conf;
-
-import org.apache.commons.configuration.ConfigurationException;
-import org.apache.commons.configuration.XMLConfiguration;
-import org.apache.commons.configuration.tree.ConfigurationNode;
-import org.apache.commons.lang.StringUtils;
-import org.apache.zeppelin.util.Util;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
-import java.net.URL;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-/**
- * Zeppelin configuration.
- *
- */
-public class ZeppelinConfiguration extends XMLConfiguration {
-  private static final String ZEPPELIN_SITE_XML = "zeppelin-site.xml";
-  private static final long serialVersionUID = 4749305895693848035L;
-  private static final Logger LOG = 
LoggerFactory.getLogger(ZeppelinConfiguration.class);
-
-  private static final String HELIUM_PACKAGE_DEFAULT_URL =
-      "https://s3.amazonaws.com/helium-package/helium.json";;
-  private static ZeppelinConfiguration conf;
-
-  public ZeppelinConfiguration(URL url) throws ConfigurationException {
-    setDelimiterParsingDisabled(true);
-    load(url);
-  }
-
-  public ZeppelinConfiguration() {
-    ConfVars[] vars = ConfVars.values();
-    for (ConfVars v : vars) {
-      if (v.getType() == ConfVars.VarType.BOOLEAN) {
-        this.setProperty(v.getVarName(), v.getBooleanValue());
-      } else if (v.getType() == ConfVars.VarType.LONG) {
-        this.setProperty(v.getVarName(), v.getLongValue());
-      } else if (v.getType() == ConfVars.VarType.INT) {
-        this.setProperty(v.getVarName(), v.getIntValue());
-      } else if (v.getType() == ConfVars.VarType.FLOAT) {
-        this.setProperty(v.getVarName(), v.getFloatValue());
-      } else if (v.getType() == ConfVars.VarType.STRING) {
-        this.setProperty(v.getVarName(), v.getStringValue());
-      } else {
-        throw new RuntimeException("Unsupported VarType");
-      }
-    }
-
-  }
-
-
-  /**
-   * Load from resource.
-   *url = ZeppelinConfiguration.class.getResource(ZEPPELIN_SITE_XML);
-   * @throws ConfigurationException
-   */
-  public static synchronized ZeppelinConfiguration create() {
-    if (conf != null) {
-      return conf;
-    }
-
-    ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
-    URL url;
-
-    url = ZeppelinConfiguration.class.getResource(ZEPPELIN_SITE_XML);
-    if (url == null) {
-      ClassLoader cl = ZeppelinConfiguration.class.getClassLoader();
-      if (cl != null) {
-        url = cl.getResource(ZEPPELIN_SITE_XML);
-      }
-    }
-    if (url == null) {
-      url = classLoader.getResource(ZEPPELIN_SITE_XML);
-    }
-
-    if (url == null) {
-      LOG.warn("Failed to load configuration, proceeding with a default");
-      conf = new ZeppelinConfiguration();
-    } else {
-      try {
-        LOG.info("Load configuration from " + url);
-        conf = new ZeppelinConfiguration(url);
-      } catch (ConfigurationException e) {
-        LOG.warn("Failed to load configuration from " + url + " proceeding 
with a default", e);
-        conf = new ZeppelinConfiguration();
-      }
-    }
-
-    LOG.info("Server Host: " + conf.getServerAddress());
-    if (conf.useSsl() == false) {
-      LOG.info("Server Port: " + conf.getServerPort());
-    } else {
-      LOG.info("Server SSL Port: " + conf.getServerSslPort());
-    }
-    LOG.info("Context Path: " + conf.getServerContextPath());
-    LOG.info("Zeppelin Version: " + Util.getVersion());
-
-    return conf;
-  }
-
-
-  private String getStringValue(String name, String d) {
-    List<ConfigurationNode> properties = getRootNode().getChildren();
-    if (properties == null || properties.isEmpty()) {
-      return d;
-    }
-    for (ConfigurationNode p : properties) {
-      if (p.getChildren("name") != null && !p.getChildren("name").isEmpty()
-          && name.equals(p.getChildren("name").get(0).getValue())) {
-        return (String) p.getChildren("value").get(0).getValue();
-      }
-    }
-    return d;
-  }
-
-  private int getIntValue(String name, int d) {
-    List<ConfigurationNode> properties = getRootNode().getChildren();
-    if (properties == null || properties.isEmpty()) {
-      return d;
-    }
-    for (ConfigurationNode p : properties) {
-      if (p.getChildren("name") != null && !p.getChildren("name").isEmpty()
-          && name.equals(p.getChildren("name").get(0).getValue())) {
-        return Integer.parseInt((String) 
p.getChildren("value").get(0).getValue());
-      }
-    }
-    return d;
-  }
-
-  private long getLongValue(String name, long d) {
-    List<ConfigurationNode> properties = getRootNode().getChildren();
-    if (properties == null || properties.isEmpty()) {
-      return d;
-    }
-    for (ConfigurationNode p : properties) {
-      if (p.getChildren("name") != null && !p.getChildren("name").isEmpty()
-          && name.equals(p.getChildren("name").get(0).getValue())) {
-        return Long.parseLong((String) 
p.getChildren("value").get(0).getValue());
-      }
-    }
-    return d;
-  }
-
-  private float getFloatValue(String name, float d) {
-    List<ConfigurationNode> properties = getRootNode().getChildren();
-    if (properties == null || properties.isEmpty()) {
-      return d;
-    }
-    for (ConfigurationNode p : properties) {
-      if (p.getChildren("name") != null && !p.getChildren("name").isEmpty()
-          && name.equals(p.getChildren("name").get(0).getValue())) {
-        return Float.parseFloat((String) 
p.getChildren("value").get(0).getValue());
-      }
-    }
-    return d;
-  }
-
-  private boolean getBooleanValue(String name, boolean d) {
-    List<ConfigurationNode> properties = getRootNode().getChildren();
-    if (properties == null || properties.isEmpty()) {
-      return d;
-    }
-    for (ConfigurationNode p : properties) {
-      if (p.getChildren("name") != null && !p.getChildren("name").isEmpty()
-          && name.equals(p.getChildren("name").get(0).getValue())) {
-        return Boolean.parseBoolean((String) 
p.getChildren("value").get(0).getValue());
-      }
-    }
-    return d;
-  }
-
-  public String getString(ConfVars c) {
-    return getString(c.name(), c.getVarName(), c.getStringValue());
-  }
-
-  public String getString(String envName, String propertyName, String 
defaultValue) {
-    if (System.getenv(envName) != null) {
-      return System.getenv(envName);
-    }
-    if (System.getProperty(propertyName) != null) {
-      return System.getProperty(propertyName);
-    }
-
-    return getStringValue(propertyName, defaultValue);
-  }
-
-  public int getInt(ConfVars c) {
-    return getInt(c.name(), c.getVarName(), c.getIntValue());
-  }
-
-  public int getInt(String envName, String propertyName, int defaultValue) {
-    if (System.getenv(envName) != null) {
-      return Integer.parseInt(System.getenv(envName));
-    }
-
-    if (System.getProperty(propertyName) != null) {
-      return Integer.parseInt(System.getProperty(propertyName));
-    }
-    return getIntValue(propertyName, defaultValue);
-  }
-
-  public long getLong(ConfVars c) {
-    return getLong(c.name(), c.getVarName(), c.getLongValue());
-  }
-
-  public long getLong(String envName, String propertyName, long defaultValue) {
-    if (System.getenv(envName) != null) {
-      return Long.parseLong(System.getenv(envName));
-    }
-
-    if (System.getProperty(propertyName) != null) {
-      return Long.parseLong(System.getProperty(propertyName));
-    }
-    return getLongValue(propertyName, defaultValue);
-  }
-
-  public float getFloat(ConfVars c) {
-    return getFloat(c.name(), c.getVarName(), c.getFloatValue());
-  }
-
-  public float getFloat(String envName, String propertyName, float 
defaultValue) {
-    if (System.getenv(envName) != null) {
-      return Float.parseFloat(System.getenv(envName));
-    }
-    if (System.getProperty(propertyName) != null) {
-      return Float.parseFloat(System.getProperty(propertyName));
-    }
-    return getFloatValue(propertyName, defaultValue);
-  }
-
-  public boolean getBoolean(ConfVars c) {
-    return getBoolean(c.name(), c.getVarName(), c.getBooleanValue());
-  }
-
-  public boolean getBoolean(String envName, String propertyName, boolean 
defaultValue) {
-    if (System.getenv(envName) != null) {
-      return Boolean.parseBoolean(System.getenv(envName));
-    }
-
-    if (System.getProperty(propertyName) != null) {
-      return Boolean.parseBoolean(System.getProperty(propertyName));
-    }
-    return getBooleanValue(propertyName, defaultValue);
-  }
-
-  public boolean useSsl() {
-    return getBoolean(ConfVars.ZEPPELIN_SSL);
-  }
-
-  public int getServerSslPort() {
-    return getInt(ConfVars.ZEPPELIN_SSL_PORT);
-  }
-
-  public boolean useClientAuth() {
-    return getBoolean(ConfVars.ZEPPELIN_SSL_CLIENT_AUTH);
-  }
-
-  public String getServerAddress() {
-    return getString(ConfVars.ZEPPELIN_ADDR);
-  }
-
-  public int getServerPort() {
-    return getInt(ConfVars.ZEPPELIN_PORT);
-  }
-
-  public String getServerContextPath() {
-    return getString(ConfVars.ZEPPELIN_SERVER_CONTEXT_PATH);
-  }
-
-  public String getKeyStorePath() {
-    String path = getString(ConfVars.ZEPPELIN_SSL_KEYSTORE_PATH);
-    if (path != null && path.startsWith("/") || isWindowsPath(path)) {
-      return path;
-    } else {
-      return getRelativeDir(
-          String.format("%s/%s",
-              getConfDir(),
-              path));
-    }
-  }
-
-  public String getKeyStoreType() {
-    return getString(ConfVars.ZEPPELIN_SSL_KEYSTORE_TYPE);
-  }
-
-  public String getKeyStorePassword() {
-    return getString(ConfVars.ZEPPELIN_SSL_KEYSTORE_PASSWORD);
-  }
-
-  public String getKeyManagerPassword() {
-    String password = getString(ConfVars.ZEPPELIN_SSL_KEY_MANAGER_PASSWORD);
-    if (password == null) {
-      return getKeyStorePassword();
-    } else {
-      return password;
-    }
-  }
-
-  public String getTrustStorePath() {
-    String path = getString(ConfVars.ZEPPELIN_SSL_TRUSTSTORE_PATH);
-    if (path == null) {
-      path = getKeyStorePath();
-    }
-    if (path != null && path.startsWith("/") || isWindowsPath(path)) {
-      return path;
-    } else {
-      return getRelativeDir(
-          String.format("%s/%s",
-              getConfDir(),
-              path));
-    }
-  }
-
-  public String getTrustStoreType() {
-    String type = getString(ConfVars.ZEPPELIN_SSL_TRUSTSTORE_TYPE);
-    if (type == null) {
-      return getKeyStoreType();
-    } else {
-      return type;
-    }
-  }
-
-  public String getTrustStorePassword() {
-    String password = getString(ConfVars.ZEPPELIN_SSL_TRUSTSTORE_PASSWORD);
-    if (password == null) {
-      return getKeyStorePassword();
-    } else {
-      return password;
-    }
-  }
-
-  public String getNotebookDir() {
-    return getString(ConfVars.ZEPPELIN_NOTEBOOK_DIR);
-  }
-
-  public String getUser() {
-    return getString(ConfVars.ZEPPELIN_NOTEBOOK_S3_USER);
-  }
-
-  public String getBucketName() {
-    return getString(ConfVars.ZEPPELIN_NOTEBOOK_S3_BUCKET);
-  }
-
-  public String getEndpoint() {
-    return getString(ConfVars.ZEPPELIN_NOTEBOOK_S3_ENDPOINT);
-  }
-
-  public String getS3KMSKeyID() {
-    return getString(ConfVars.ZEPPELIN_NOTEBOOK_S3_KMS_KEY_ID);
-  }
-
-  public String getS3KMSKeyRegion() {
-    return getString(ConfVars.ZEPPELIN_NOTEBOOK_S3_KMS_KEY_REGION);
-  }
-
-  public String getS3EncryptionMaterialsProviderClass() {
-    return getString(ConfVars.ZEPPELIN_NOTEBOOK_S3_EMP);
-  }
-
-  public boolean isS3ServerSideEncryption() {
-    return getBoolean(ConfVars.ZEPPELIN_NOTEBOOK_S3_SSE);
-  }
-
-  public String getMongoUri() {
-    return getString(ConfVars.ZEPPELIN_NOTEBOOK_MONGO_URI);
-  }
-
-  public String getMongoDatabase() {
-    return getString(ConfVars.ZEPPELIN_NOTEBOOK_MONGO_DATABASE);
-  }
-
-  public String getMongoCollection() {
-    return getString(ConfVars.ZEPPELIN_NOTEBOOK_MONGO_COLLECTION);
-  }
-
-  public boolean getMongoAutoimport() {
-    return getBoolean(ConfVars.ZEPPELIN_NOTEBOOK_MONGO_AUTOIMPORT);
-  }
-
-  public String getInterpreterListPath() {
-    return getRelativeDir(String.format("%s/interpreter-list", getConfDir()));
-  }
-
-  public String getInterpreterDir() {
-    return getRelativeDir(ConfVars.ZEPPELIN_INTERPRETER_DIR);
-  }
-
-  public String getInterpreterJson() {
-    return getString(ConfVars.ZEPPELIN_INTERPRETER_JSON);
-  }
-
-  public String getInterpreterSettingPath() {
-    return getRelativeDir(String.format("%s/interpreter.json", getConfDir()));
-  }
-
-  public String getHeliumConfPath() {
-    return getRelativeDir(String.format("%s/helium.json", getConfDir()));
-  }
-
-  public String getHeliumRegistry() {
-    return getRelativeDir(ConfVars.ZEPPELIN_HELIUM_REGISTRY);
-  }
-
-  public String getHeliumNodeInstallerUrl() {
-    return getString(ConfVars.ZEPPELIN_HELIUM_NODE_INSTALLER_URL);
-  }
-
-  public String getHeliumNpmInstallerUrl() {
-    return getString(ConfVars.ZEPPELIN_HELIUM_NPM_INSTALLER_URL);
-  }
-
-  public String getHeliumYarnInstallerUrl() {
-    return getString(ConfVars.ZEPPELIN_HELIUM_YARNPKG_INSTALLER_URL);
-  }
-
-  public String getNotebookAuthorizationPath() {
-    return getRelativeDir(String.format("%s/notebook-authorization.json", 
getConfDir()));
-  }
-
-  public Boolean credentialsPersist() {
-    return getBoolean(ConfVars.ZEPPELIN_CREDENTIALS_PERSIST);
-  }
-
-  public String getCredentialsEncryptKey() {
-    return getString(ConfVars.ZEPPELIN_CREDENTIALS_ENCRYPT_KEY);
-  }
-
-  public String getCredentialsPath() {
-    return getRelativeDir(String.format("%s/credentials.json", getConfDir()));
-  }
-
-  public String getShiroPath() {
-    String shiroPath = getRelativeDir(String.format("%s/shiro.ini", 
getConfDir()));
-    return new File(shiroPath).exists() ? shiroPath : StringUtils.EMPTY;
-  }
-
-  public String getInterpreterRemoteRunnerPath() {
-    return getRelativeDir(ConfVars.ZEPPELIN_INTERPRETER_REMOTE_RUNNER);
-  }
-
-  public String getInterpreterLocalRepoPath() {
-    return getRelativeDir(ConfVars.ZEPPELIN_INTERPRETER_LOCALREPO);
-  }
-
-  public String getInterpreterMvnRepoPath() {
-    return getString(ConfVars.ZEPPELIN_INTERPRETER_DEP_MVNREPO);
-  }
-
-  public String getRelativeDir(ConfVars c) {
-    return getRelativeDir(getString(c));
-  }
-
-  public String getRelativeDir(String path) {
-    if (path != null && path.startsWith("/") || isWindowsPath(path)) {
-      return path;
-    } else {
-      return getString(ConfVars.ZEPPELIN_HOME) + "/" + path;
-    }
-  }
-
-  public String getCallbackPortRange() {
-    return getString(ConfVars.ZEPPELIN_INTERPRETER_CALLBACK_PORTRANGE);
-  }
-
-  public boolean isWindowsPath(String path){
-    return path.matches("^[A-Za-z]:\\\\.*");
-  }
-
-  public boolean isAnonymousAllowed() {
-    return getBoolean(ConfVars.ZEPPELIN_ANONYMOUS_ALLOWED);
-  }
-
-  public boolean isNotebokPublic() {
-    return getBoolean(ConfVars.ZEPPELIN_NOTEBOOK_PUBLIC);
-  }
-
-  public String getConfDir() {
-    return getRelativeDir(ConfVars.ZEPPELIN_CONF_DIR);
-  }
-
-  public List<String> getAllowedOrigins()
-  {
-    if (getString(ConfVars.ZEPPELIN_ALLOWED_ORIGINS).isEmpty()) {
-      return Arrays.asList(new String[0]);
-    }
-
-    return 
Arrays.asList(getString(ConfVars.ZEPPELIN_ALLOWED_ORIGINS).toLowerCase().split(","));
-  }
-
-  public String getWebsocketMaxTextMessageSize() {
-    return getString(ConfVars.ZEPPELIN_WEBSOCKET_MAX_TEXT_MESSAGE_SIZE);
-  }
-
-  public String getJettyName() {
-    return getString(ConfVars.ZEPPELIN_SERVER_JETTY_NAME);
-  }
-
-
-  public String getXFrameOptions() {
-    return getString(ConfVars.ZEPPELIN_SERVER_XFRAME_OPTIONS);
-  }
-
-  public String getXxssProtection() {
-    return getString(ConfVars.ZEPPELIN_SERVER_X_XSS_PROTECTION);
-  }
-
-  public String getStrictTransport() {
-    return getString(ConfVars.ZEPPELIN_SERVER_STRICT_TRANSPORT);
-  }
-
-
-  public Map<String, String> dumpConfigurations(ZeppelinConfiguration conf,
-                                                ConfigurationKeyPredicate 
predicate) {
-    Map<String, String> configurations = new HashMap<>();
-
-    for (ConfVars v : ConfVars.values()) {
-      String key = v.getVarName();
-
-      if (!predicate.apply(key)) {
-        continue;
-      }
-
-      ConfVars.VarType type = v.getType();
-      Object value = null;
-      if (type == ConfVars.VarType.BOOLEAN) {
-        value = conf.getBoolean(v);
-      } else if (type == ConfVars.VarType.LONG) {
-        value = conf.getLong(v);
-      } else if (type == ConfVars.VarType.INT) {
-        value = conf.getInt(v);
-      } else if (type == ConfVars.VarType.FLOAT) {
-        value = conf.getFloat(v);
-      } else if (type == ConfVars.VarType.STRING) {
-        value = conf.getString(v);
-      }
-
-      if (value != null) {
-        configurations.put(key, value.toString());
-      }
-    }
-    return configurations;
-  }
-
-  /**
-   * Predication whether key/value pair should be included or not
-   */
-  public interface ConfigurationKeyPredicate {
-    boolean apply(String key);
-  }
-
-  /**
-   * Wrapper class.
-   */
-  public static enum ConfVars {
-    ZEPPELIN_HOME("zeppelin.home", "./"),
-    ZEPPELIN_ADDR("zeppelin.server.addr", "0.0.0.0"),
-    ZEPPELIN_PORT("zeppelin.server.port", 8080),
-    ZEPPELIN_SERVER_CONTEXT_PATH("zeppelin.server.context.path", "/"),
-    ZEPPELIN_SSL("zeppelin.ssl", false),
-    ZEPPELIN_SSL_PORT("zeppelin.server.ssl.port", 8443),
-    ZEPPELIN_SSL_CLIENT_AUTH("zeppelin.ssl.client.auth", false),
-    ZEPPELIN_SSL_KEYSTORE_PATH("zeppelin.ssl.keystore.path", "keystore"),
-    ZEPPELIN_SSL_KEYSTORE_TYPE("zeppelin.ssl.keystore.type", "JKS"),
-    ZEPPELIN_SSL_KEYSTORE_PASSWORD("zeppelin.ssl.keystore.password", ""),
-    ZEPPELIN_SSL_KEY_MANAGER_PASSWORD("zeppelin.ssl.key.manager.password", 
null),
-    ZEPPELIN_SSL_TRUSTSTORE_PATH("zeppelin.ssl.truststore.path", null),
-    ZEPPELIN_SSL_TRUSTSTORE_TYPE("zeppelin.ssl.truststore.type", null),
-    ZEPPELIN_SSL_TRUSTSTORE_PASSWORD("zeppelin.ssl.truststore.password", null),
-    ZEPPELIN_WAR("zeppelin.war", "zeppelin-web/dist"),
-    ZEPPELIN_WAR_TEMPDIR("zeppelin.war.tempdir", "webapps"),
-    ZEPPELIN_INTERPRETERS("zeppelin.interpreters", 
"org.apache.zeppelin.spark.SparkInterpreter,"
-        + "org.apache.zeppelin.spark.PySparkInterpreter,"
-        + "org.apache.zeppelin.rinterpreter.RRepl,"
-        + "org.apache.zeppelin.rinterpreter.KnitR,"
-        + "org.apache.zeppelin.spark.SparkRInterpreter,"
-        + "org.apache.zeppelin.spark.SparkSqlInterpreter,"
-        + "org.apache.zeppelin.spark.DepInterpreter,"
-        + "org.apache.zeppelin.markdown.Markdown,"
-        + "org.apache.zeppelin.angular.AngularInterpreter,"
-        + "org.apache.zeppelin.shell.ShellInterpreter,"
-        + "org.apache.zeppelin.livy.LivySparkInterpreter,"
-        + "org.apache.zeppelin.livy.LivySparkSQLInterpreter,"
-        + "org.apache.zeppelin.livy.LivyPySparkInterpreter,"
-        + "org.apache.zeppelin.livy.LivyPySpark3Interpreter,"
-        + "org.apache.zeppelin.livy.LivySparkRInterpreter,"
-        + "org.apache.zeppelin.alluxio.AlluxioInterpreter,"
-        + "org.apache.zeppelin.file.HDFSFileInterpreter,"
-        + "org.apache.zeppelin.pig.PigInterpreter,"
-        + "org.apache.zeppelin.pig.PigQueryInterpreter,"
-        + "org.apache.zeppelin.flink.FlinkInterpreter,"
-        + "org.apache.zeppelin.python.PythonInterpreter,"
-        + "org.apache.zeppelin.python.PythonInterpreterPandasSql,"
-        + "org.apache.zeppelin.python.PythonCondaInterpreter,"
-        + "org.apache.zeppelin.python.PythonDockerInterpreter,"
-        + "org.apache.zeppelin.ignite.IgniteInterpreter,"
-        + "org.apache.zeppelin.ignite.IgniteSqlInterpreter,"
-        + "org.apache.zeppelin.lens.LensInterpreter,"
-        + "org.apache.zeppelin.cassandra.CassandraInterpreter,"
-        + "org.apache.zeppelin.geode.GeodeOqlInterpreter,"
-        + "org.apache.zeppelin.kylin.KylinInterpreter,"
-        + "org.apache.zeppelin.elasticsearch.ElasticsearchInterpreter,"
-        + "org.apache.zeppelin.scalding.ScaldingInterpreter,"
-        + "org.apache.zeppelin.jdbc.JDBCInterpreter,"
-        + "org.apache.zeppelin.hbase.HbaseInterpreter,"
-        + "org.apache.zeppelin.bigquery.BigQueryInterpreter,"
-        + "org.apache.zeppelin.beam.BeamInterpreter,"
-        + "org.apache.zeppelin.scio.ScioInterpreter,"
-        + "org.apache.zeppelin.groovy.GroovyInterpreter,"
-        + "org.apache.zeppelin.neo4j.Neo4jCypherInterpreter"
-        ),
-    ZEPPELIN_INTERPRETER_JSON("zeppelin.interpreter.setting", 
"interpreter-setting.json"),
-    ZEPPELIN_INTERPRETER_DIR("zeppelin.interpreter.dir", "interpreter"),
-    ZEPPELIN_INTERPRETER_LOCALREPO("zeppelin.interpreter.localRepo", 
"local-repo"),
-    ZEPPELIN_INTERPRETER_DEP_MVNREPO("zeppelin.interpreter.dep.mvnRepo",
-        "http://repo1.maven.org/maven2/";),
-    
ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT("zeppelin.interpreter.connect.timeout", 
30000),
-    ZEPPELIN_INTERPRETER_MAX_POOL_SIZE("zeppelin.interpreter.max.poolsize", 
10),
-    ZEPPELIN_INTERPRETER_GROUP_ORDER("zeppelin.interpreter.group.order", 
"spark,md,angular,sh,"
-        + 
"livy,alluxio,file,psql,flink,python,ignite,lens,cassandra,geode,kylin,elasticsearch,"
-        + "scalding,jdbc,hbase,bigquery,beam,pig,scio,groovy,neo4j"),
-    ZEPPELIN_INTERPRETER_OUTPUT_LIMIT("zeppelin.interpreter.output.limit", 
1024 * 100),
-    ZEPPELIN_ENCODING("zeppelin.encoding", "UTF-8"),
-    ZEPPELIN_NOTEBOOK_DIR("zeppelin.notebook.dir", "notebook"),
-    // use specified notebook (id) as homescreen
-    ZEPPELIN_NOTEBOOK_HOMESCREEN("zeppelin.notebook.homescreen", null),
-    // whether homescreen notebook will be hidden from notebook list or not
-    ZEPPELIN_NOTEBOOK_HOMESCREEN_HIDE("zeppelin.notebook.homescreen.hide", 
false),
-    ZEPPELIN_NOTEBOOK_S3_BUCKET("zeppelin.notebook.s3.bucket", "zeppelin"),
-    ZEPPELIN_NOTEBOOK_S3_ENDPOINT("zeppelin.notebook.s3.endpoint", 
"s3.amazonaws.com"),
-    ZEPPELIN_NOTEBOOK_S3_USER("zeppelin.notebook.s3.user", "user"),
-    
ZEPPELIN_NOTEBOOK_S3_EMP("zeppelin.notebook.s3.encryptionMaterialsProvider", 
null),
-    ZEPPELIN_NOTEBOOK_S3_KMS_KEY_ID("zeppelin.notebook.s3.kmsKeyID", null),
-    ZEPPELIN_NOTEBOOK_S3_KMS_KEY_REGION("zeppelin.notebook.s3.kmsKeyRegion", 
null),
-    ZEPPELIN_NOTEBOOK_S3_SSE("zeppelin.notebook.s3.sse", false),
-    
ZEPPELIN_NOTEBOOK_AZURE_CONNECTION_STRING("zeppelin.notebook.azure.connectionString",
 null),
-    ZEPPELIN_NOTEBOOK_AZURE_SHARE("zeppelin.notebook.azure.share", "zeppelin"),
-    ZEPPELIN_NOTEBOOK_AZURE_USER("zeppelin.notebook.azure.user", "user"),
-    ZEPPELIN_NOTEBOOK_MONGO_DATABASE("zeppelin.notebook.mongo.database", 
"zeppelin"),
-    ZEPPELIN_NOTEBOOK_MONGO_COLLECTION("zeppelin.notebook.mongo.collection", 
"notes"),
-    ZEPPELIN_NOTEBOOK_MONGO_URI("zeppelin.notebook.mongo.uri", 
"mongodb://localhost"),
-    ZEPPELIN_NOTEBOOK_MONGO_AUTOIMPORT("zeppelin.notebook.mongo.autoimport", 
false),
-    ZEPPELIN_NOTEBOOK_STORAGE("zeppelin.notebook.storage",
-        "org.apache.zeppelin.notebook.repo.GitNotebookRepo"),
-    ZEPPELIN_NOTEBOOK_ONE_WAY_SYNC("zeppelin.notebook.one.way.sync", false),
-    // whether by default note is public or private
-    ZEPPELIN_NOTEBOOK_PUBLIC("zeppelin.notebook.public", true),
-    ZEPPELIN_INTERPRETER_REMOTE_RUNNER("zeppelin.interpreter.remoterunner",
-        System.getProperty("os.name")
-                .startsWith("Windows") ? "bin/interpreter.cmd" : 
"bin/interpreter.sh"),
-    // Decide when new note is created, interpreter settings will be binded 
automatically or not.
-    
ZEPPELIN_NOTEBOOK_AUTO_INTERPRETER_BINDING("zeppelin.notebook.autoInterpreterBinding",
 true),
-    ZEPPELIN_CONF_DIR("zeppelin.conf.dir", "conf"),
-    ZEPPELIN_DEP_LOCALREPO("zeppelin.dep.localrepo", "local-repo"),
-    ZEPPELIN_HELIUM_REGISTRY("zeppelin.helium.registry", "helium," + 
HELIUM_PACKAGE_DEFAULT_URL),
-    ZEPPELIN_HELIUM_NODE_INSTALLER_URL("zeppelin.helium.node.installer.url",
-            "https://nodejs.org/dist/";),
-    ZEPPELIN_HELIUM_NPM_INSTALLER_URL("zeppelin.helium.npm.installer.url",
-            "http://registry.npmjs.org/";),
-    
ZEPPELIN_HELIUM_YARNPKG_INSTALLER_URL("zeppelin.helium.yarnpkg.installer.url",
-            "https://github.com/yarnpkg/yarn/releases/download/";),
-    // Allows a way to specify a ',' separated list of allowed origins for 
rest and websockets
-    // i.e. http://localhost:8080
-    ZEPPELIN_ALLOWED_ORIGINS("zeppelin.server.allowed.origins", "*"),
-    ZEPPELIN_ANONYMOUS_ALLOWED("zeppelin.anonymous.allowed", true),
-    ZEPPELIN_CREDENTIALS_PERSIST("zeppelin.credentials.persist", true),
-    ZEPPELIN_CREDENTIALS_ENCRYPT_KEY("zeppelin.credentials.encryptKey", null),
-    
ZEPPELIN_WEBSOCKET_MAX_TEXT_MESSAGE_SIZE("zeppelin.websocket.max.text.message.size",
 "1024000"),
-    ZEPPELIN_SERVER_DEFAULT_DIR_ALLOWED("zeppelin.server.default.dir.allowed", 
false),
-    ZEPPELIN_SERVER_XFRAME_OPTIONS("zeppelin.server.xframe.options", 
"SAMEORIGIN"),
-    ZEPPELIN_SERVER_JETTY_NAME("zeppelin.server.jetty.name", null),
-    ZEPPELIN_SERVER_STRICT_TRANSPORT("zeppelin.server.strict.transport", 
"max-age=631138519"),
-    ZEPPELIN_SERVER_X_XSS_PROTECTION("zeppelin.server.xxss.protection", "1"),
-
-    ZEPPELIN_SERVER_KERBEROS_KEYTAB("zeppelin.server.kerberos.keytab", ""),
-    ZEPPELIN_SERVER_KERBEROS_PRINCIPAL("zeppelin.server.kerberos.principal", 
""),
-
-    
ZEPPELIN_INTERPRETER_CALLBACK_PORTRANGE("zeppelin.interpreter.callback.portRange",
 ":");
-
-    private String varName;
-    @SuppressWarnings("rawtypes")
-    private Class varClass;
-    private String stringValue;
-    private VarType type;
-    private int intValue;
-    private float floatValue;
-    private boolean booleanValue;
-    private long longValue;
-
-
-    ConfVars(String varName, String varValue) {
-      this.varName = varName;
-      this.varClass = String.class;
-      this.stringValue = varValue;
-      this.intValue = -1;
-      this.floatValue = -1;
-      this.longValue = -1;
-      this.booleanValue = false;
-      this.type = VarType.STRING;
-    }
-
-    ConfVars(String varName, int intValue) {
-      this.varName = varName;
-      this.varClass = Integer.class;
-      this.stringValue = null;
-      this.intValue = intValue;
-      this.floatValue = -1;
-      this.longValue = -1;
-      this.booleanValue = false;
-      this.type = VarType.INT;
-    }
-
-    ConfVars(String varName, long longValue) {
-      this.varName = varName;
-      this.varClass = Integer.class;
-      this.stringValue = null;
-      this.intValue = -1;
-      this.floatValue = -1;
-      this.longValue = longValue;
-      this.booleanValue = false;
-      this.type = VarType.LONG;
-    }
-
-    ConfVars(String varName, float floatValue) {
-      this.varName = varName;
-      this.varClass = Float.class;
-      this.stringValue = null;
-      this.intValue = -1;
-      this.longValue = -1;
-      this.floatValue = floatValue;
-      this.booleanValue = false;
-      this.type = VarType.FLOAT;
-    }
-
-    ConfVars(String varName, boolean booleanValue) {
-      this.varName = varName;
-      this.varClass = Boolean.class;
-      this.stringValue = null;
-      this.intValue = -1;
-      this.longValue = -1;
-      this.floatValue = -1;
-      this.booleanValue = booleanValue;
-      this.type = VarType.BOOLEAN;
-    }
-
-    public String getVarName() {
-      return varName;
-    }
-
-    @SuppressWarnings("rawtypes")
-    public Class getVarClass() {
-      return varClass;
-    }
-
-    public int getIntValue() {
-      return intValue;
-    }
-
-    public long getLongValue() {
-      return longValue;
-    }
-
-    public float getFloatValue() {
-      return floatValue;
-    }
-
-    public String getStringValue() {
-      return stringValue;
-    }
-
-    public boolean getBooleanValue() {
-      return booleanValue;
-    }
-
-    public VarType getType() {
-      return type;
-    }
-
-    enum VarType {
-      STRING {
-        @Override
-        void checkType(String value) throws Exception {}
-      },
-      INT {
-        @Override
-        void checkType(String value) throws Exception {
-          Integer.valueOf(value);
-        }
-      },
-      LONG {
-        @Override
-        void checkType(String value) throws Exception {
-          Long.valueOf(value);
-        }
-      },
-      FLOAT {
-        @Override
-        void checkType(String value) throws Exception {
-          Float.valueOf(value);
-        }
-      },
-      BOOLEAN {
-        @Override
-        void checkType(String value) throws Exception {
-          Boolean.valueOf(value);
-        }
-      };
-
-      boolean isType(String value) {
-        try {
-          checkType(value);
-        } catch (Exception e) {
-          LOG.error("Exception in ZeppelinConfiguration while isType", e);
-          return false;
-        }
-        return true;
-      }
-
-      String typeString() {
-        return name().toUpperCase();
-      }
-
-      abstract void checkType(String value) throws Exception;
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java
 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java
index f020919..7233239 100644
--- 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java
+++ 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java
@@ -77,7 +77,7 @@ public class InterpreterFactory {
           return interpreter;
         }
       }
-      throw new InterpreterException(replName + " interpreter not found");
+      return null;
 
     } else {
       // first assume replName is 'name' of interpreter. ('groupName' is 
ommitted)

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSetting.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSetting.java
 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSetting.java
index 5af01dc..a82d5bf 100644
--- 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSetting.java
+++ 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSetting.java
@@ -18,7 +18,6 @@
 package org.apache.zeppelin.interpreter;
 
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
 import com.google.gson.JsonArray;
@@ -34,19 +33,22 @@ import org.apache.zeppelin.dep.DependencyResolver;
 import org.apache.zeppelin.display.AngularObjectRegistry;
 import org.apache.zeppelin.display.AngularObjectRegistryListener;
 import org.apache.zeppelin.helium.ApplicationEventListener;
+import org.apache.zeppelin.interpreter.launcher.InterpreterLaunchContext;
+import org.apache.zeppelin.interpreter.launcher.InterpreterLauncher;
+import org.apache.zeppelin.interpreter.launcher.ShellScriptLauncher;
+import org.apache.zeppelin.interpreter.launcher.SparkInterpreterLauncher;
 import org.apache.zeppelin.interpreter.remote.RemoteAngularObjectRegistry;
 import org.apache.zeppelin.interpreter.remote.RemoteInterpreter;
-import org.apache.zeppelin.interpreter.remote.RemoteInterpreterManagedProcess;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterEventPoller;
 import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess;
 import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcessListener;
-import org.apache.zeppelin.interpreter.remote.RemoteInterpreterRunningProcess;
 import org.apache.zeppelin.interpreter.remote.RemoteInterpreterUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.io.FileNotFoundException;
-import java.io.FilenameFilter;
+import java.io.IOException;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
 import java.net.URL;
@@ -58,7 +60,6 @@ import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
-import java.util.NoSuchElementException;
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -132,6 +133,10 @@ public class InterpreterSetting {
 
   private transient ZeppelinConfiguration conf = new ZeppelinConfiguration();
 
+  // TODO(zjffdu) ShellScriptLauncher is the only launcher implemention for 
now. It could be other
+  // launcher in future when we have other launcher implementation. e.g. third 
party launcher
+  // service like livy
+  private transient InterpreterLauncher launcher;
   
///////////////////////////////////////////////////////////////////////////////////////////
 
 
@@ -243,6 +248,7 @@ public class InterpreterSetting {
   }
 
   void postProcessing() {
+//    createLauncher();
     this.status = Status.READY;
   }
 
@@ -266,6 +272,14 @@ public class InterpreterSetting {
     this.conf = o.getConf();
   }
 
+  private void createLauncher() {
+    if (group.equals("spark")) {
+      this.launcher = new SparkInterpreterLauncher(this.conf);
+    } else {
+      this.launcher = new ShellScriptLauncher(this.conf);
+    }
+  }
+
   public AngularObjectRegistryListener getAngularObjectRegistryListener() {
     return angularObjectRegistryListener;
   }
@@ -626,152 +640,17 @@ public class InterpreterSetting {
     }
     return interpreters;
   }
-  
-  RemoteInterpreterProcess createInterpreterProcess() {
-    RemoteInterpreterProcess remoteInterpreterProcess = null;
-    int connectTimeout =
-        
conf.getInt(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT);
-    String localRepoPath = conf.getInterpreterLocalRepoPath() + "/" + id;
-    if (option.isExistingProcess()) {
-      // TODO(zjffdu) remove the existing process approach seems no one is 
using this.
-      // use the existing process
-      remoteInterpreterProcess = new RemoteInterpreterRunningProcess(
-          connectTimeout,
-          remoteInterpreterProcessListener,
-          appEventListener,
-          option.getHost(),
-          option.getPort());
-    } else {
-      // create new remote process
-      remoteInterpreterProcess = new RemoteInterpreterManagedProcess(
-          interpreterRunner != null ? interpreterRunner.getPath() :
-              conf.getInterpreterRemoteRunnerPath(), 
conf.getCallbackPortRange(),
-          interpreterDir, localRepoPath,
-          getEnvFromInterpreterProperty(), connectTimeout,
-          remoteInterpreterProcessListener, appEventListener, group);
-    }
-    return remoteInterpreterProcess;
-  }
-
-  private boolean isSparkConf(String key, String value) {
-    return !StringUtils.isEmpty(key) && key.startsWith("spark.") && 
!StringUtils.isEmpty(value);
-  }
-
-  private Map<String, String> getEnvFromInterpreterProperty() {
-    Map<String, String> env = new HashMap<String, String>();
-    Properties javaProperties = getJavaProperties();
-    Properties sparkProperties = new Properties();
-    String sparkMaster = getSparkMaster();
-    for (String key : javaProperties.stringPropertyNames()) {
-      if (RemoteInterpreterUtils.isEnvString(key)) {
-        env.put(key, javaProperties.getProperty(key));
-      }
-      if (isSparkConf(key, javaProperties.getProperty(key))) {
-        sparkProperties.setProperty(key, 
toShellFormat(javaProperties.getProperty(key)));
-      }
-    }
-
-    setupPropertiesForPySpark(sparkProperties);
-    setupPropertiesForSparkR(sparkProperties, System.getenv("SPARK_HOME"));
-    if (isYarnMode() && getDeployMode().equals("cluster")) {
-      env.put("SPARK_YARN_CLUSTER", "true");
-    }
-
-    StringBuilder sparkConfBuilder = new StringBuilder();
-    if (sparkMaster != null) {
-      sparkConfBuilder.append(" --master " + sparkMaster);
-    }
-    if (isYarnMode() && getDeployMode().equals("cluster")) {
-      sparkConfBuilder.append(" --files " + conf.getConfDir() + 
"/log4j_yarn_cluster.properties");
-    }
-    for (String name : sparkProperties.stringPropertyNames()) {
-      sparkConfBuilder.append(" --conf " + name + "=" + 
sparkProperties.getProperty(name));
-    }
-
-    env.put("ZEPPELIN_SPARK_CONF", sparkConfBuilder.toString());
-    LOGGER.debug("getEnvFromInterpreterProperty: " + env);
-    return env;
-  }
-
-  private void setupPropertiesForPySpark(Properties sparkProperties) {
-    if (isYarnMode()) {
-      sparkProperties.setProperty("spark.yarn.isPython", "true");
-    }
-  }
 
-  private void mergeSparkProperty(Properties sparkProperties, String 
propertyName,
-                                  String propertyValue) {
-    if (sparkProperties.containsKey(propertyName)) {
-      String oldPropertyValue = sparkProperties.getProperty(propertyName);
-      sparkProperties.setProperty(propertyName, oldPropertyValue + "," + 
propertyValue);
-    } else {
-      sparkProperties.setProperty(propertyName, propertyValue);
-    }
-  }
-
-  private void setupPropertiesForSparkR(Properties sparkProperties,
-                                        String sparkHome) {
-    File sparkRBasePath = null;
-    if (sparkHome == null) {
-      if (!getSparkMaster().startsWith("local")) {
-        throw new RuntimeException("SPARK_HOME is not specified for non-local 
mode");
-      }
-      String zeppelinHome = 
conf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_HOME);
-      sparkRBasePath = new File(zeppelinHome,
-          "interpreter" + File.separator + "spark" + File.separator + "R");
-    } else {
-      sparkRBasePath = new File(sparkHome, "R" + File.separator + "lib");
-    }
-
-    File sparkRPath = new File(sparkRBasePath, "sparkr.zip");
-    if (sparkRPath.exists() && sparkRPath.isFile()) {
-      mergeSparkProperty(sparkProperties, "spark.yarn.dist.archives", 
sparkRPath.getAbsolutePath());
-    } else {
-      LOGGER.warn("sparkr.zip is not found, SparkR may not work.");
-    }
-  }
-
-  private String getSparkMaster() {
-    String master = getJavaProperties().getProperty("master");
-    if (master == null) {
-      master = getJavaProperties().getProperty("spark.master", "local[*]");
-    }
-    return master;
-  }
-
-  private String getDeployMode() {
-    String master = getSparkMaster();
-    if (master.equals("yarn-client")) {
-      return "client";
-    } else if (master.equals("yarn-cluster")) {
-      return "cluster";
-    } else if (master.startsWith("local")) {
-      return "client";
-    } else {
-      String deployMode = 
getJavaProperties().getProperty("spark.submit.deployMode");
-      if (deployMode == null) {
-        throw new RuntimeException("master is set as yarn, but 
spark.submit.deployMode " +
-            "is not specified");
-      }
-      if (!deployMode.equals("client") && !deployMode.equals("cluster")) {
-        throw new RuntimeException("Invalid value for spark.submit.deployMode: 
" + deployMode);
-      }
-      return deployMode;
-    }
-  }
-
-  private boolean isYarnMode() {
-    return getSparkMaster().startsWith("yarn");
-  }
-
-  private String toShellFormat(String value) {
-    if (value.contains("\'") && value.contains("\"")) {
-      throw new RuntimeException("Spark property value could not contain both 
\" and '");
-    } else if (value.contains("\'")) {
-      return "\"" + value + "\"";
-    } else {
-      return "\'" + value + "\'";
+  synchronized RemoteInterpreterProcess createInterpreterProcess() throws 
IOException {
+    if (launcher == null) {
+      createLauncher();
     }
+    InterpreterLaunchContext launchContext = new
+        InterpreterLaunchContext(getJavaProperties(), option, 
interpreterRunner, id, name);
+    RemoteInterpreterProcess process = (RemoteInterpreterProcess) 
launcher.launch(launchContext);
+    process.setRemoteInterpreterEventPoller(
+        new RemoteInterpreterEventPoller(remoteInterpreterProcessListener, 
appEventListener));
+    return process;
   }
 
   private List<Interpreter> getOrCreateSession(String user, String noteId) {
@@ -815,8 +694,7 @@ public class InterpreterSetting {
     return null;
   }
 
-  private ManagedInterpreterGroup createInterpreterGroup(String groupId)
-      throws InterpreterException {
+  private ManagedInterpreterGroup createInterpreterGroup(String groupId) {
     AngularObjectRegistry angularObjectRegistry;
     ManagedInterpreterGroup interpreterGroup = new 
ManagedInterpreterGroup(groupId, this);
     angularObjectRegistry =
@@ -938,7 +816,8 @@ public class InterpreterSetting {
           );
           newProperties.put(key, property);
         } else {
-          throw new RuntimeException("Can not convert this type of property: " 
+ value.getClass());
+          throw new RuntimeException("Can not convert this type of property: " 
+
+              value.getClass());
         }
       }
       return newProperties;

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java
 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java
index 9dfce21..f34195d 100644
--- 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java
+++ 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java
@@ -744,11 +744,12 @@ public class InterpreterSettingManager {
   }
 
   /**
-   * Change interpreter property and restart
+   * Change interpreter properties and restart
    */
   public void setPropertyAndRestart(String id, InterpreterOption option,
                                     Map<String, InterpreterProperty> 
properties,
-                                    List<Dependency> dependencies) throws 
IOException {
+                                    List<Dependency> dependencies)
+      throws InterpreterException, IOException {
     synchronized (interpreterSettings) {
       InterpreterSetting intpSetting = interpreterSettings.get(id);
       if (intpSetting != null) {
@@ -761,7 +762,7 @@ public class InterpreterSettingManager {
           saveToFile();
         } catch (Exception e) {
           loadFromFile();
-          throw e;
+          throw new IOException(e);
         }
       } else {
         throw new InterpreterException("Interpreter setting id " + id + " not 
found");
@@ -770,7 +771,7 @@ public class InterpreterSettingManager {
   }
 
   // restart in note page
-  public void restart(String settingId, String noteId, String user) {
+  public void restart(String settingId, String noteId, String user) throws 
InterpreterException {
     InterpreterSetting intpSetting = interpreterSettings.get(settingId);
     Preconditions.checkNotNull(intpSetting);
     synchronized (interpreterSettings) {
@@ -794,7 +795,7 @@ public class InterpreterSettingManager {
     }
   }
 
-  public void restart(String id) {
+  public void restart(String id) throws InterpreterException {
     restart(id, "", "anonymous");
   }
 

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java
 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java
index 1d7d916..ff9cb1c 100644
--- 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java
+++ 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java
@@ -25,6 +25,7 @@ import org.apache.zeppelin.scheduler.SchedulerFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.util.Collection;
 import java.util.List;
 
@@ -52,7 +53,7 @@ public class ManagedInterpreterGroup extends InterpreterGroup 
{
     return interpreterSetting;
   }
 
-  public synchronized RemoteInterpreterProcess getOrCreateInterpreterProcess() 
{
+  public synchronized RemoteInterpreterProcess getOrCreateInterpreterProcess() 
throws IOException {
     if (remoteInterpreterProcess == null) {
       LOGGER.info("Create InterperterProcess for InterpreterGroup: " + 
getId());
       remoteInterpreterProcess = interpreterSetting.createInterpreterProcess();
@@ -112,7 +113,11 @@ public class ManagedInterpreterGroup extends 
InterpreterGroup {
         LOGGER.info("Job " + job.getJobName() + " aborted ");
       }
 
-      interpreter.close();
+      try {
+        interpreter.close();
+      } catch (InterpreterException e) {
+        LOGGER.warn("Fail to close interpreter " + interpreter.getClassName(), 
e);
+      }
       //TODO(zjffdu) move the close of schedule to Interpreter
       if (null != scheduler) {
         SchedulerFactory.singleton().removeScheduler(scheduler.getName());

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/ShellScriptLauncher.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/ShellScriptLauncher.java
 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/ShellScriptLauncher.java
new file mode 100644
index 0000000..f419967
--- /dev/null
+++ 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/ShellScriptLauncher.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.zeppelin.interpreter.launcher;
+
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.InterpreterOption;
+import org.apache.zeppelin.interpreter.InterpreterRunner;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterManagedProcess;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterRunningProcess;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * Interpreter Launcher which use shell script to launch the interpreter 
process.
+ *
+ */
+public class ShellScriptLauncher extends InterpreterLauncher {
+
+  private static final Logger LOGGER = 
LoggerFactory.getLogger(ShellScriptLauncher.class);
+
+  public ShellScriptLauncher(ZeppelinConfiguration zConf) {
+    super(zConf);
+  }
+
+  @Override
+  public InterpreterClient launch(InterpreterLaunchContext context) {
+    LOGGER.info("Launching Interpreter: " + context.getInterpreterGroupName());
+    this.properties = context.getProperties();
+    InterpreterOption option = context.getOption();
+    InterpreterRunner runner = context.getRunner();
+    String groupName = context.getInterpreterGroupName();
+
+    int connectTimeout =
+        
zConf.getInt(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT);
+    if (option.isExistingProcess()) {
+      return new RemoteInterpreterRunningProcess(
+          connectTimeout,
+          option.getHost(),
+          option.getPort());
+    } else {
+      // create new remote process
+      String localRepoPath = zConf.getInterpreterLocalRepoPath() + "/"
+          + context.getInterpreterGroupId();
+      return new RemoteInterpreterManagedProcess(
+          runner != null ? runner.getPath() : 
zConf.getInterpreterRemoteRunnerPath(),
+          zConf.getCallbackPortRange(),
+          zConf.getInterpreterDir() + "/" + groupName, localRepoPath,
+          buildEnvFromProperties(), connectTimeout, groupName);
+    }
+  }
+
+  protected Map<String, String> buildEnvFromProperties() {
+    Map<String, String> env = new HashMap<>();
+    for (Object key : properties.keySet()) {
+      if (RemoteInterpreterUtils.isEnvString((String) key)) {
+        env.put((String) key, properties.getProperty((String) key));
+      }
+    }
+    return env;
+  }
+}

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncher.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncher.java
 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncher.java
new file mode 100644
index 0000000..32a0530
--- /dev/null
+++ 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncher.java
@@ -0,0 +1,205 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.interpreter.launcher;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * Spark specific launcher.
+ */
+public class SparkInterpreterLauncher extends ShellScriptLauncher {
+
+  private static final Logger LOGGER = 
LoggerFactory.getLogger(SparkInterpreterLauncher.class);
+
+  public SparkInterpreterLauncher(ZeppelinConfiguration zConf) {
+    super(zConf);
+  }
+
+  @Override
+  protected Map<String, String> buildEnvFromProperties() {
+    Map<String, String> env = new HashMap<String, String>();
+    Properties sparkProperties = new Properties();
+    String sparkMaster = getSparkMaster(properties);
+    for (String key : properties.stringPropertyNames()) {
+      if (RemoteInterpreterUtils.isEnvString(key)) {
+        env.put(key, properties.getProperty(key));
+      }
+      if (isSparkConf(key, properties.getProperty(key))) {
+        sparkProperties.setProperty(key, 
toShellFormat(properties.getProperty(key)));
+      }
+    }
+
+    setupPropertiesForPySpark(sparkProperties);
+    setupPropertiesForSparkR(sparkProperties);
+    if (isYarnMode() && getDeployMode().equals("cluster")) {
+      env.put("ZEPPELIN_SPARK_YARN_CLUSTER", "true");
+    }
+
+    StringBuilder sparkConfBuilder = new StringBuilder();
+    if (sparkMaster != null) {
+      sparkConfBuilder.append(" --master " + sparkMaster);
+    }
+    if (isYarnMode() && getDeployMode().equals("cluster")) {
+      sparkConfBuilder.append(" --files " + zConf.getConfDir() + 
"/log4j_yarn_cluster.properties");
+    }
+    for (String name : sparkProperties.stringPropertyNames()) {
+      sparkConfBuilder.append(" --conf " + name + "=" + 
sparkProperties.getProperty(name));
+    }
+
+    env.put("ZEPPELIN_SPARK_CONF", sparkConfBuilder.toString());
+
+    // set these env in the order of
+    // 1. interpreter-setting
+    // 2. zeppelin-env.sh
+    // It is encouraged to set env in interpreter setting, but just for 
backward compatability,
+    // we also fallback to zeppelin-env.sh if it is not specified in 
interpreter setting.
+    for (String envName : new String[]{"SPARK_HOME", "SPARK_CONF_DIR", 
"HADOOP_CONF_DIR"})  {
+      String envValue = getEnv(envName);
+      if (envValue != null) {
+        env.put(envName, envValue);
+      }
+    }
+    LOGGER.debug("buildEnvFromProperties: " + env);
+    return env;
+
+  }
+
+
+  /**
+   * get environmental variable in the following order
+   *
+   * 1. interpreter setting
+   * 2. zeppelin-env.sh
+   *
+   */
+  private String getEnv(String envName) {
+    String env = properties.getProperty(envName);
+    if (env == null) {
+      env = System.getenv(envName);
+    }
+    return env;
+  }
+
+  private boolean isSparkConf(String key, String value) {
+    return !StringUtils.isEmpty(key) && key.startsWith("spark.") && 
!StringUtils.isEmpty(value);
+  }
+
+  private void setupPropertiesForPySpark(Properties sparkProperties) {
+    if (isYarnMode()) {
+      sparkProperties.setProperty("spark.yarn.isPython", "true");
+    }
+  }
+
+  private void mergeSparkProperty(Properties sparkProperties, String 
propertyName,
+                                  String propertyValue) {
+    if (sparkProperties.containsKey(propertyName)) {
+      String oldPropertyValue = sparkProperties.getProperty(propertyName);
+      sparkProperties.setProperty(propertyName, oldPropertyValue + "," + 
propertyValue);
+    } else {
+      sparkProperties.setProperty(propertyName, propertyValue);
+    }
+  }
+
+  private void setupPropertiesForSparkR(Properties sparkProperties) {
+    String sparkHome = getEnv("SPARK_HOME");
+    File sparkRBasePath = null;
+    if (sparkHome == null) {
+      if (!getSparkMaster(properties).startsWith("local")) {
+        throw new RuntimeException("SPARK_HOME is not specified in 
interpreter-setting" +
+            " for non-local mode, if you specify it in zeppelin-env.sh, please 
move that into " +
+            " interpreter setting");
+      }
+      String zeppelinHome = 
zConf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_HOME);
+      sparkRBasePath = new File(zeppelinHome,
+          "interpreter" + File.separator + "spark" + File.separator + "R");
+    } else {
+      sparkRBasePath = new File(sparkHome, "R" + File.separator + "lib");
+    }
+
+    File sparkRPath = new File(sparkRBasePath, "sparkr.zip");
+    if (sparkRPath.exists() && sparkRPath.isFile()) {
+      mergeSparkProperty(sparkProperties, "spark.yarn.dist.archives", 
sparkRPath.getAbsolutePath());
+    } else {
+      LOGGER.warn("sparkr.zip is not found, SparkR may not work.");
+    }
+  }
+
+  /**
+   * Order to look for spark master
+   * 1. master in interpreter setting
+   * 2. spark.master interpreter setting
+   * 3. use local[*]
+   * @param properties
+   * @return
+   */
+  private String getSparkMaster(Properties properties) {
+    String master = properties.getProperty("master");
+    if (master == null) {
+      master = properties.getProperty("spark.master");
+      if (master == null) {
+        master = "local[*]";
+      }
+    }
+    return master;
+  }
+
+  private String getDeployMode() {
+    String master = getSparkMaster(properties);
+    if (master.equals("yarn-client")) {
+      return "client";
+    } else if (master.equals("yarn-cluster")) {
+      return "cluster";
+    } else if (master.startsWith("local")) {
+      return "client";
+    } else {
+      String deployMode = properties.getProperty("spark.submit.deployMode");
+      if (deployMode == null) {
+        throw new RuntimeException("master is set as yarn, but 
spark.submit.deployMode " +
+            "is not specified");
+      }
+      if (!deployMode.equals("client") && !deployMode.equals("cluster")) {
+        throw new RuntimeException("Invalid value for spark.submit.deployMode: 
" + deployMode);
+      }
+      return deployMode;
+    }
+  }
+
+  private boolean isYarnMode() {
+    return getSparkMaster(properties).startsWith("yarn");
+  }
+
+  private String toShellFormat(String value) {
+    if (value.contains("\'") && value.contains("\"")) {
+      throw new RuntimeException("Spark property value could not contain both 
\" and '");
+    } else if (value.contains("\'")) {
+      return "\"" + value + "\"";
+    } else {
+      return "\'" + value + "\'";
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/InterpreterContextRunnerPool.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/InterpreterContextRunnerPool.java
 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/InterpreterContextRunnerPool.java
index 064abd5..7653824 100644
--- 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/InterpreterContextRunnerPool.java
+++ 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/InterpreterContextRunnerPool.java
@@ -82,7 +82,7 @@ public class InterpreterContextRunnerPool {
         }
       }
 
-      throw new InterpreterException("Can not run paragraph " + paragraphId + 
" on " + noteId);
+      throw new RuntimeException("Can not run paragraph " + paragraphId + " on 
" + noteId);
     }
   }
 }

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java
 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java
index 54bf9e1..b479799 100644
--- 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java
+++ 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java
@@ -28,6 +28,7 @@ import org.apache.zeppelin.display.Input;
 import org.apache.zeppelin.interpreter.Interpreter;
 import org.apache.zeppelin.interpreter.InterpreterContext;
 import org.apache.zeppelin.interpreter.InterpreterContextRunner;
+import org.apache.zeppelin.interpreter.InterpreterException;
 import org.apache.zeppelin.interpreter.InterpreterResult;
 import org.apache.zeppelin.interpreter.ManagedInterpreterGroup;
 import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
@@ -42,6 +43,7 @@ import org.apache.zeppelin.scheduler.SchedulerFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -90,7 +92,7 @@ public class RemoteInterpreter extends Interpreter {
     return this.sessionId;
   }
 
-  public synchronized RemoteInterpreterProcess getOrCreateInterpreterProcess() 
{
+  public synchronized RemoteInterpreterProcess getOrCreateInterpreterProcess() 
throws IOException {
     if (this.interpreterProcess != null) {
       return this.interpreterProcess;
     }
@@ -113,7 +115,7 @@ public class RemoteInterpreter extends Interpreter {
   }
 
   @Override
-  public void open() {
+  public void open() throws InterpreterException {
     synchronized (this) {
       if (!isOpened) {
         // create all the interpreters of the same session first, then Open 
the internal interpreter
@@ -123,7 +125,11 @@ public class RemoteInterpreter extends Interpreter {
         // also see method 
Interpreter.getInterpreterInTheSameSessionByClassName
         for (Interpreter interpreter : getInterpreterGroup()
                                         .getOrCreateSession(userName, 
sessionId)) {
-          ((RemoteInterpreter) interpreter).internal_create();
+          try {
+            ((RemoteInterpreter) interpreter).internal_create();
+          } catch (IOException e) {
+            throw new InterpreterException(e);
+          }
         }
 
         interpreterProcess.callRemoteFunction(new 
RemoteInterpreterProcess.RemoteFunction<Void>() {
@@ -147,7 +153,7 @@ public class RemoteInterpreter extends Interpreter {
     }
   }
 
-  private void internal_create() {
+  private void internal_create() throws IOException {
     synchronized (this) {
       if (!isCreated) {
         RemoteInterpreterProcess interpreterProcess = 
getOrCreateInterpreterProcess();
@@ -156,7 +162,7 @@ public class RemoteInterpreter extends Interpreter {
           public Void call(Client client) throws Exception {
             LOGGER.info("Create RemoteInterpreter {}", getClassName());
             client.createInterpreter(getInterpreterGroup().getId(), sessionId,
-                className, (Map) property, userName);
+                className, (Map) getProperties(), userName);
             return null;
           }
         });
@@ -167,9 +173,14 @@ public class RemoteInterpreter extends Interpreter {
 
 
   @Override
-  public void close() {
+  public void close() throws InterpreterException {
     if (isOpened) {
-      RemoteInterpreterProcess interpreterProcess = 
getOrCreateInterpreterProcess();
+      RemoteInterpreterProcess interpreterProcess = null;
+      try {
+        interpreterProcess = getOrCreateInterpreterProcess();
+      } catch (IOException e) {
+        throw new InterpreterException(e);
+      }
       interpreterProcess.callRemoteFunction(new 
RemoteInterpreterProcess.RemoteFunction<Void>() {
         @Override
         public Void call(Client client) throws Exception {
@@ -184,13 +195,19 @@ public class RemoteInterpreter extends Interpreter {
   }
 
   @Override
-  public InterpreterResult interpret(final String st, final InterpreterContext 
context) {
+  public InterpreterResult interpret(final String st, final InterpreterContext 
context)
+      throws InterpreterException {
     if (LOGGER.isDebugEnabled()) {
       LOGGER.debug("st:\n{}", st);
     }
 
     final FormType form = getFormType();
-    RemoteInterpreterProcess interpreterProcess = 
getOrCreateInterpreterProcess();
+    RemoteInterpreterProcess interpreterProcess = null;
+    try {
+      interpreterProcess = getOrCreateInterpreterProcess();
+    } catch (IOException e) {
+      throw new InterpreterException(e);
+    }
     InterpreterContextRunnerPool interpreterContextRunnerPool = 
interpreterProcess
         .getInterpreterContextRunnerPool();
     List<InterpreterContextRunner> runners = context.getRunners();
@@ -238,12 +255,17 @@ public class RemoteInterpreter extends Interpreter {
   }
 
   @Override
-  public void cancel(final InterpreterContext context) {
+  public void cancel(final InterpreterContext context) throws 
InterpreterException {
     if (!isOpened) {
       LOGGER.warn("Cancel is called when RemoterInterpreter is not opened for 
" + className);
       return;
     }
-    RemoteInterpreterProcess interpreterProcess = 
getOrCreateInterpreterProcess();
+    RemoteInterpreterProcess interpreterProcess = null;
+    try {
+      interpreterProcess = getOrCreateInterpreterProcess();
+    } catch (IOException e) {
+      throw new InterpreterException(e);
+    }
     interpreterProcess.callRemoteFunction(new 
RemoteInterpreterProcess.RemoteFunction<Void>() {
       @Override
       public Void call(Client client) throws Exception {
@@ -254,7 +276,7 @@ public class RemoteInterpreter extends Interpreter {
   }
 
   @Override
-  public FormType getFormType() {
+  public FormType getFormType() throws InterpreterException {
     if (formType != null) {
       return formType;
     }
@@ -265,7 +287,12 @@ public class RemoteInterpreter extends Interpreter {
         open();
       }
     }
-    RemoteInterpreterProcess interpreterProcess = 
getOrCreateInterpreterProcess();
+    RemoteInterpreterProcess interpreterProcess = null;
+    try {
+      interpreterProcess = getOrCreateInterpreterProcess();
+    } catch (IOException e) {
+      throw new InterpreterException(e);
+    }
     FormType type = interpreterProcess.callRemoteFunction(
         new RemoteInterpreterProcess.RemoteFunction<FormType>() {
           @Override
@@ -277,13 +304,19 @@ public class RemoteInterpreter extends Interpreter {
     return type;
   }
 
+
   @Override
-  public int getProgress(final InterpreterContext context) {
+  public int getProgress(final InterpreterContext context) throws 
InterpreterException {
     if (!isOpened) {
       LOGGER.warn("getProgress is called when RemoterInterpreter is not opened 
for " + className);
       return 0;
     }
-    RemoteInterpreterProcess interpreterProcess = 
getOrCreateInterpreterProcess();
+    RemoteInterpreterProcess interpreterProcess = null;
+    try {
+      interpreterProcess = getOrCreateInterpreterProcess();
+    } catch (IOException e) {
+      throw new InterpreterException(e);
+    }
     return interpreterProcess.callRemoteFunction(
         new RemoteInterpreterProcess.RemoteFunction<Integer>() {
           @Override
@@ -296,12 +329,18 @@ public class RemoteInterpreter extends Interpreter {
 
   @Override
   public List<InterpreterCompletion> completion(final String buf, final int 
cursor,
-                                                final InterpreterContext 
interpreterContext) {
+                                                final InterpreterContext 
interpreterContext)
+      throws InterpreterException {
     if (!isOpened) {
       LOGGER.warn("completion is called when RemoterInterpreter is not opened 
for " + className);
       return new ArrayList<>();
     }
-    RemoteInterpreterProcess interpreterProcess = 
getOrCreateInterpreterProcess();
+    RemoteInterpreterProcess interpreterProcess = null;
+    try {
+      interpreterProcess = getOrCreateInterpreterProcess();
+    } catch (IOException e) {
+      throw new InterpreterException(e);
+    }
     return interpreterProcess.callRemoteFunction(
         new 
RemoteInterpreterProcess.RemoteFunction<List<InterpreterCompletion>>() {
           @Override
@@ -317,7 +356,12 @@ public class RemoteInterpreter extends Interpreter {
       LOGGER.warn("getStatus is called when RemoteInterpreter is not opened 
for " + className);
       return Job.Status.UNKNOWN.name();
     }
-    RemoteInterpreterProcess interpreterProcess = 
getOrCreateInterpreterProcess();
+    RemoteInterpreterProcess interpreterProcess = null;
+    try {
+      interpreterProcess = getOrCreateInterpreterProcess();
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
     return interpreterProcess.callRemoteFunction(
         new RemoteInterpreterProcess.RemoteFunction<String>() {
           @Override
@@ -331,7 +375,7 @@ public class RemoteInterpreter extends Interpreter {
   @Override
   public Scheduler getScheduler() {
     int maxConcurrency = Integer.parseInt(
-        property.getProperty("zeppelin.interpreter.max.poolsize",
+        getProperty("zeppelin.interpreter.max.poolsize",
             
ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_MAX_POOL_SIZE.getIntValue() 
+ ""));
 
     Scheduler s = new RemoteScheduler(

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterManagedProcess.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterManagedProcess.java
 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterManagedProcess.java
index d21a962..6e26e58 100644
--- 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterManagedProcess.java
+++ 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterManagedProcess.java
@@ -17,6 +17,7 @@
 
 package org.apache.zeppelin.interpreter.remote;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.commons.exec.CommandLine;
 import org.apache.commons.exec.DefaultExecutor;
 import org.apache.commons.exec.ExecuteException;
@@ -73,11 +74,8 @@ public class RemoteInterpreterManagedProcess extends 
RemoteInterpreterProcess
       String localRepoDir,
       Map<String, String> env,
       int connectTimeout,
-      RemoteInterpreterProcessListener listener,
-      ApplicationEventListener appListener,
       String interpreterGroupName) {
-    super(new RemoteInterpreterEventPoller(listener, appListener),
-        connectTimeout);
+    super(connectTimeout);
     this.interpreterRunner = intpRunner;
     this.portRange = portRange;
     this.env = env;
@@ -86,23 +84,6 @@ public class RemoteInterpreterManagedProcess extends 
RemoteInterpreterProcess
     this.interpreterGroupName = interpreterGroupName;
   }
 
-  RemoteInterpreterManagedProcess(String intpRunner,
-                                  String intpDir,
-                                  String localRepoDir,
-                                  Map<String, String> env,
-                                  RemoteInterpreterEventPoller 
remoteInterpreterEventPoller,
-                                  int connectTimeout,
-                                  String interpreterGroupName) {
-    super(remoteInterpreterEventPoller,
-        connectTimeout);
-    this.interpreterRunner = intpRunner;
-    this.portRange = ":";
-    this.env = env;
-    this.interpreterDir = intpDir;
-    this.localRepoDir = localRepoDir;
-    this.interpreterGroupName = interpreterGroupName;
-  }
-
   @Override
   public String getHost() {
     return "localhost";
@@ -124,7 +105,7 @@ public class RemoteInterpreterManagedProcess extends 
RemoteInterpreterProcess
       callbackHost = RemoteInterpreterUtils.findAvailableHostAddress();
       callbackPort = 
RemoteInterpreterUtils.findRandomAvailablePortOnAllLocalInterfaces();
     } catch (IOException e1) {
-      throw new InterpreterException(e1);
+      throw new RuntimeException(e1);
     }
 
     logger.info("Thrift server for callback will start. Port: {}", 
callbackPort);
@@ -206,7 +187,7 @@ public class RemoteInterpreterManagedProcess extends 
RemoteInterpreterProcess
       executor.execute(cmdLine, procEnv, this);
     } catch (IOException e) {
       running.set(false);
-      throw new InterpreterException(e);
+      throw new RuntimeException(e);
     }
 
     try {
@@ -217,7 +198,7 @@ public class RemoteInterpreterManagedProcess extends 
RemoteInterpreterProcess
       }
       if (!running.get()) {
         callbackServer.stop();
-        throw new InterpreterException("Cannot run interpreter");
+        throw new RuntimeException(new String(cmdOut.toByteArray()));
       }
     } catch (InterruptedException e) {
       logger.error("Remote interpreter is not accessible");
@@ -227,7 +208,7 @@ public class RemoteInterpreterManagedProcess extends 
RemoteInterpreterProcess
 
   public void stop() {
     // shutdown EventPoller first.
-    this.remoteInterpreterEventPoller.shutdown();
+    this.getRemoteInterpreterEventPoller().shutdown();
     if (callbackServer.isServing()) {
       callbackServer.stop();
     }
@@ -266,6 +247,31 @@ public class RemoteInterpreterManagedProcess extends 
RemoteInterpreterProcess
     running.set(false);
   }
 
+  @VisibleForTesting
+  public Map<String, String> getEnv() {
+    return env;
+  }
+
+  @VisibleForTesting
+  public String getLocalRepoDir() {
+    return localRepoDir;
+  }
+
+  @VisibleForTesting
+  public String getInterpreterDir() {
+    return interpreterDir;
+  }
+
+  @VisibleForTesting
+  public String getInterpreterGroupName() {
+    return interpreterGroupName;
+  }
+
+  @VisibleForTesting
+  public String getInterpreterRunner() {
+    return interpreterRunner;
+  }
+
   public boolean isRunning() {
     return running.get();
   }

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java
 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java
index e45f15b..88cc489 100644
--- 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java
+++ 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java
@@ -21,6 +21,7 @@ import org.apache.commons.pool2.impl.GenericObjectPool;
 import org.apache.thrift.TException;
 import org.apache.zeppelin.helium.ApplicationEventListener;
 import org.apache.zeppelin.interpreter.InterpreterException;
+import org.apache.zeppelin.interpreter.launcher.InterpreterClient;
 import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterService.Client;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -28,27 +29,17 @@ import org.slf4j.LoggerFactory;
 /**
  * Abstract class for interpreter process
  */
-public abstract class RemoteInterpreterProcess {
+public abstract class RemoteInterpreterProcess implements InterpreterClient {
   private static final Logger logger = 
LoggerFactory.getLogger(RemoteInterpreterProcess.class);
 
   private GenericObjectPool<Client> clientPool;
-  protected final RemoteInterpreterEventPoller remoteInterpreterEventPoller;
+  private RemoteInterpreterEventPoller remoteInterpreterEventPoller;
   private final InterpreterContextRunnerPool interpreterContextRunnerPool;
   private int connectTimeout;
 
   public RemoteInterpreterProcess(
-      int connectTimeout,
-      RemoteInterpreterProcessListener listener,
-      ApplicationEventListener appListener) {
-    this(new RemoteInterpreterEventPoller(listener, appListener),
-        connectTimeout);
-    this.remoteInterpreterEventPoller.setInterpreterProcess(this);
-  }
-
-  RemoteInterpreterProcess(RemoteInterpreterEventPoller 
remoteInterpreterEventPoller,
-                           int connectTimeout) {
+      int connectTimeout) {
     this.interpreterContextRunnerPool = new InterpreterContextRunnerPool();
-    this.remoteInterpreterEventPoller = remoteInterpreterEventPoller;
     this.connectTimeout = connectTimeout;
   }
 
@@ -56,6 +47,10 @@ public abstract class RemoteInterpreterProcess {
     return remoteInterpreterEventPoller;
   }
 
+  public void setRemoteInterpreterEventPoller(RemoteInterpreterEventPoller 
eventPoller) {
+    this.remoteInterpreterEventPoller = eventPoller;
+  }
+
   public abstract String getHost();
   public abstract int getPort();
   public abstract void start(String userName, Boolean isUserImpersonate);
@@ -147,9 +142,9 @@ public abstract class RemoteInterpreterProcess {
       }
     } catch (TException e) {
       broken = true;
-      throw new InterpreterException(e);
+      throw new RuntimeException(e);
     } catch (Exception e1) {
-      throw new InterpreterException(e1);
+      throw new RuntimeException(e1);
     } finally {
       if (client != null) {
         releaseClient(client, broken);

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterRunningProcess.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterRunningProcess.java
 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterRunningProcess.java
index bb176be..d8715a0 100644
--- 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterRunningProcess.java
+++ 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterRunningProcess.java
@@ -30,12 +30,10 @@ public class RemoteInterpreterRunningProcess extends 
RemoteInterpreterProcess {
 
   public RemoteInterpreterRunningProcess(
       int connectTimeout,
-      RemoteInterpreterProcessListener listener,
-      ApplicationEventListener appListener,
       String host,
       int port
   ) {
-    super(connectTimeout, listener, appListener);
+    super(connectTimeout);
     this.host = host;
     this.port = port;
   }

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java
index 03c5046..b5dda67 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java
@@ -627,7 +627,7 @@ public class Note implements ParagraphJobListener, 
JsonSerializable {
     if (intp == null) {
       String intpExceptionMsg =
           p.getJobName() + "'s Interpreter " + requiredReplName + " not found";
-      InterpreterException intpException = new 
InterpreterException(intpExceptionMsg);
+      RuntimeException intpException = new RuntimeException(intpExceptionMsg);
       InterpreterResult intpResult =
           new InterpreterResult(InterpreterResult.Code.ERROR, 
intpException.getMessage());
       p.setReturn(intpResult, intpException);

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java
index 4652fcd..77fd04c 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java
@@ -892,7 +892,11 @@ public class Notebook implements NoteEventListener {
       if (releaseResource) {
         for (InterpreterSetting setting : 
notebook.getInterpreterSettingManager()
             .getInterpreterSettings(note.getId())) {
-          notebook.getInterpreterSettingManager().restart(setting.getId());
+          try {
+            notebook.getInterpreterSettingManager().restart(setting.getId());
+          } catch (InterpreterException e) {
+            logger.error("Fail to restart interpreter: " + setting.getId(), e);
+          }
         }
       }
     }

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java 
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java
index 68ce794..701943a 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java
@@ -312,15 +312,14 @@ public class Paragraph extends Job implements Cloneable, 
JsonSerializable {
     String replName = getRequiredReplName(trimmedBuffer);
 
     String body = getScriptBody(trimmedBuffer);
-    Interpreter repl = getRepl(replName);
-    if (repl == null) {
-      return null;
-    }
-
     InterpreterContext interpreterContext = 
getInterpreterContextWithoutRunner(null);
 
-    List completion = repl.completion(body, cursor, interpreterContext);
-    return completion;
+    try {
+      Interpreter repl = getRepl(replName);
+      return repl.completion(body, cursor, interpreterContext);
+    } catch (InterpreterException e) {
+      throw new RuntimeException("Fail to get completion", e);
+    }
   }
 
   public int calculateCursorPosition(String buffer, String trimmedBuffer, int 
cursor) {
@@ -362,11 +361,15 @@ public class Paragraph extends Job implements Cloneable, 
JsonSerializable {
   @Override
   public int progress() {
     String replName = getRequiredReplName();
-    Interpreter repl = getRepl(replName);
-    if (repl != null) {
+
+    try {
+      Interpreter repl = getRepl(replName);
+      if (repl == null) {
+        return 0;
+      }
       return repl.getProgress(getInterpreterContext(null));
-    } else {
-      return 0;
+    } catch (InterpreterException e) {
+      throw new RuntimeException("Fail to get progress", e);
     }
   }
 
@@ -494,10 +497,8 @@ public class Paragraph extends Job implements Cloneable, 
JsonSerializable {
   protected boolean jobAbort() {
     Interpreter repl = getRepl(getRequiredReplName());
     if (repl == null) {
-      // when interpreters are already destroyed
       return true;
     }
-
     Scheduler scheduler = repl.getScheduler();
     if (scheduler == null) {
       return true;
@@ -507,7 +508,11 @@ public class Paragraph extends Job implements Cloneable, 
JsonSerializable {
     if (job != null) {
       job.setStatus(Status.ABORT);
     } else {
-      repl.cancel(getInterpreterContextWithoutRunner(null));
+      try {
+        repl.cancel(getInterpreterContextWithoutRunner(null));
+      } catch (InterpreterException e) {
+        throw new RuntimeException(e);
+      }
     }
     return true;
   }
@@ -738,12 +743,7 @@ public class Paragraph extends Job implements Cloneable, 
JsonSerializable {
   }
 
   private boolean isValidInterpreter(String replName) {
-    try {
-      return factory.getInterpreter(user, note.getId(), replName) != null;
-    } catch (InterpreterException e) {
-      // ignore this exception, it would be recaught when running paragraph.
-      return false;
-    }
+    return factory.getInterpreter(user, note.getId(), replName) != null;
   }
 
   public void updateRuntimeInfos(String label, String tooltip, Map<String, 
String> infos,

http://git-wip-us.apache.org/repos/asf/zeppelin/blob/9812e26b/zeppelin-zengine/src/test/java/org/apache/zeppelin/helium/HeliumApplicationFactoryTest.java
----------------------------------------------------------------------
diff --git 
a/zeppelin-zengine/src/test/java/org/apache/zeppelin/helium/HeliumApplicationFactoryTest.java
 
b/zeppelin-zengine/src/test/java/org/apache/zeppelin/helium/HeliumApplicationFactoryTest.java
index 8f3e615..f23d433 100644
--- 
a/zeppelin-zengine/src/test/java/org/apache/zeppelin/helium/HeliumApplicationFactoryTest.java
+++ 
b/zeppelin-zengine/src/test/java/org/apache/zeppelin/helium/HeliumApplicationFactoryTest.java
@@ -19,6 +19,7 @@ package org.apache.zeppelin.helium;
 import org.apache.zeppelin.conf.ZeppelinConfiguration;
 import org.apache.zeppelin.interpreter.AbstractInterpreterTest;
 import org.apache.zeppelin.interpreter.Interpreter;
+import org.apache.zeppelin.interpreter.InterpreterException;
 import org.apache.zeppelin.interpreter.InterpreterResultMessage;
 import org.apache.zeppelin.interpreter.InterpreterSetting;
 import org.apache.zeppelin.notebook.ApplicationState;
@@ -241,7 +242,7 @@ public class HeliumApplicationFactoryTest extends 
AbstractInterpreterTest implem
 
 
   @Test
-  public void testUnloadOnInterpreterRestart() throws IOException {
+  public void testUnloadOnInterpreterRestart() throws IOException, 
InterpreterException {
     // given
     HeliumPackage pkg1 = new HeliumPackage(HeliumType.APPLICATION,
         "name1",

Reply via email to