You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zeppelin.apache.org by jo...@apache.org on 2026/03/06 08:49:13 UTC
(zeppelin) branch ZEPPELIN-6400-remove-zepconf-from-interpreter created (now a3f6c6730b)
This is an automated email from the ASF dual-hosted git repository.
jongyoul pushed a change to branch ZEPPELIN-6400-remove-zepconf-from-interpreter
in repository https://gitbox.apache.org/repos/asf/zeppelin.git
at a3f6c6730b [ZEPPELIN-6400] Remove ZeppelinConfiguration dependency from zeppelin-interpreter module
This branch includes the following new commits:
new a3f6c6730b [ZEPPELIN-6400] Remove ZeppelinConfiguration dependency from zeppelin-interpreter module
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(zeppelin) 01/01: [ZEPPELIN-6400] Remove ZeppelinConfiguration dependency from zeppelin-interpreter module
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jongyoul pushed a commit to branch ZEPPELIN-6400-remove-zepconf-from-interpreter
in repository https://gitbox.apache.org/repos/asf/zeppelin.git
commit a3f6c6730b3706876cd3e109a6bd2847713be574
Author: Jongyoul Lee <jo...@gmail.com>
AuthorDate: Fri Mar 6 17:48:53 2026 +0900
[ZEPPELIN-6400] Remove ZeppelinConfiguration dependency from zeppelin-interpreter module
Move ZeppelinConfiguration to zeppelin-zengine so it is no longer
included in the shaded interpreter JAR. This prevents the Maven shade
plugin from corrupting config string literals, which caused classpath-
order-dependent configuration loading failures.
Key changes:
- Replace ZeppelinConfiguration usage in zeppelin-interpreter with
Properties-based configuration
- Update InterpreterLauncher, LifecycleManager, RecoveryStorage,
DependencyResolver to accept Properties instead of ZeppelinConfiguration
- Update all launcher plugins (Docker, K8s, YARN, Flink) accordingly
- Move ZeppelinConfiguration.java and ZeppelinLocationStrategy.java
from zeppelin-interpreter to zeppelin-zengine
- Update callers in zeppelin-zengine, zeppelin-server, flink, and
markdown interpreters
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
---
.../apache/zeppelin/flink/FlinkInterpreter.java | 5 +--
.../zeppelin/flink/FlinkScala212Interpreter.scala | 6 +--
.../zeppelin/flink/FlinkScalaInterpreter.scala | 11 +++--
.../apache/zeppelin/markdown/FlexmarkParser.java | 5 +--
.../org/apache/zeppelin/markdown/Markdown.java | 4 +-
.../zeppelin/markdown/FlexmarkParserTest.java | 2 -
.../zeppelin/dep/AbstractDependencyResolver.java | 23 +++++-----
.../main/java/org/apache/zeppelin/dep/Booter.java | 5 +--
.../apache/zeppelin/dep/DependencyResolver.java | 10 ++---
.../apache/zeppelin/interpreter/Interpreter.java | 6 ---
.../zeppelin/interpreter/InterpreterOption.java | 12 +++---
.../zeppelin/interpreter/LazyOpenInterpreter.java | 6 ---
.../zeppelin/interpreter/LifecycleManager.java | 9 ++--
.../interpreter/launcher/InterpreterLauncher.java | 43 ++++++++++++-------
.../lifecycle/NullLifecycleManager.java | 7 ++--
.../lifecycle/TimeoutLifecycleManager.java | 21 ++++++----
.../interpreter/recovery/RecoveryStorage.java | 7 +---
.../remote/RemoteInterpreterServer.java | 34 ++++++++-------
.../zeppelin/scheduler/SchedulerFactory.java | 16 +++++--
.../zeppelin/conf/ZeppelinConfigurationTest.java | 49 ----------------------
.../java/org/apache/zeppelin/dep/BooterTest.java | 5 +--
.../zeppelin/dep/DependencyResolverTest.java | 3 +-
.../zeppelin/helium/ApplicationLoaderTest.java | 3 +-
.../remote/RemoteInterpreterServerTest.java | 2 +-
.../launcher/DockerInterpreterLauncher.java | 16 +++----
.../launcher/DockerInterpreterProcess.java | 31 +++++++-------
.../launcher/DockerInterpreterProcessTest.java | 29 ++++++++-----
.../launcher/FlinkInterpreterLauncher.java | 7 ++--
.../launcher/K8sStandardInterpreterLauncher.java | 22 +++++-----
.../zeppelin/interpreter/launcher/K8sUtils.java | 10 ++---
.../K8sStandardInterpreterLauncherTest.java | 12 ++++--
.../launcher/YarnInterpreterLauncher.java | 8 ++--
.../launcher/YarnRemoteInterpreterProcess.java | 15 ++++---
.../zeppelin/service/InterpreterService.java | 8 +++-
.../zeppelin/service/InterpreterServiceTest.java | 2 +-
.../zeppelin/conf/ZeppelinConfiguration.java | 0
.../zeppelin/conf/ZeppelinLocationStrategy.java | 0
.../zeppelin/interpreter/InterpreterSetting.java | 4 +-
.../interpreter/InterpreterSettingManager.java | 5 ++-
.../interpreter/install/InstallInterpreter.java | 3 +-
.../launcher/SparkInterpreterLauncher.java | 33 ++++++++-------
.../launcher/StandardInterpreterLauncher.java | 20 +++++----
.../recovery/FileSystemRecoveryStorage.java | 4 +-
.../interpreter/recovery/LocalRecoveryStorage.java | 7 +++-
.../interpreter/recovery/NullRecoveryStorage.java | 2 +-
.../interpreter/remote/RemoteInterpreter.java | 1 -
.../org/apache/zeppelin/plugin/PluginManager.java | 20 +++++++--
.../interpreter/InterpreterSettingTest.java | 2 +-
.../launcher/SparkInterpreterLauncherTest.java | 41 +++++++++++-------
.../launcher/StandardInterpreterLauncherTest.java | 20 ++++++---
50 files changed, 320 insertions(+), 296 deletions(-)
diff --git a/flink/flink-scala-2.12/src/main/java/org/apache/zeppelin/flink/FlinkInterpreter.java b/flink/flink-scala-2.12/src/main/java/org/apache/zeppelin/flink/FlinkInterpreter.java
index 244d00f49c..fa37e0fb30 100644
--- a/flink/flink-scala-2.12/src/main/java/org/apache/zeppelin/flink/FlinkInterpreter.java
+++ b/flink/flink-scala-2.12/src/main/java/org/apache/zeppelin/flink/FlinkInterpreter.java
@@ -22,7 +22,6 @@ import org.apache.flink.api.scala.ExecutionEnvironment;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableEnvironment;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.*;
import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
import org.slf4j.Logger;
@@ -91,8 +90,8 @@ public class FlinkInterpreter extends Interpreter {
Class<?> clazz = Class.forName(innerIntpClassName);
return (FlinkScalaInterpreter)
- clazz.getConstructor(Properties.class, ClassLoader.class, ZeppelinConfiguration.class)
- .newInstance(getProperties(), flinkScalaClassLoader, zConf);
+ clazz.getConstructor(Properties.class, ClassLoader.class)
+ .newInstance(getProperties(), flinkScalaClassLoader);
}
@Override
diff --git a/flink/flink-scala-2.12/src/main/scala/org/apache/zeppelin/flink/FlinkScala212Interpreter.scala b/flink/flink-scala-2.12/src/main/scala/org/apache/zeppelin/flink/FlinkScala212Interpreter.scala
index 070daf6a2b..b7af4ad3b1 100644
--- a/flink/flink-scala-2.12/src/main/scala/org/apache/zeppelin/flink/FlinkScala212Interpreter.scala
+++ b/flink/flink-scala-2.12/src/main/scala/org/apache/zeppelin/flink/FlinkScala212Interpreter.scala
@@ -24,15 +24,13 @@ import java.util.Properties
import org.apache.zeppelin.interpreter.InterpreterContext
import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion
-import org.apache.zeppelin.conf.ZeppelinConfiguration
import scala.tools.nsc.Settings
import scala.tools.nsc.interpreter.{IMain, JPrintWriter}
class FlinkScala212Interpreter(override val properties: Properties,
- override val flinkScalaClassLoader: ClassLoader,
- override val zConf: ZeppelinConfiguration)
- extends FlinkScalaInterpreter(properties, flinkScalaClassLoader, zConf) {
+ override val flinkScalaClassLoader: ClassLoader)
+ extends FlinkScalaInterpreter(properties, flinkScalaClassLoader) {
override def completion(buf: String,
cursor: Int,
diff --git a/flink/flink-scala-2.12/src/main/scala/org/apache/zeppelin/flink/FlinkScalaInterpreter.scala b/flink/flink-scala-2.12/src/main/scala/org/apache/zeppelin/flink/FlinkScalaInterpreter.scala
index c2ceb60360..a11003b8bc 100644
--- a/flink/flink-scala-2.12/src/main/scala/org/apache/zeppelin/flink/FlinkScalaInterpreter.scala
+++ b/flink/flink-scala-2.12/src/main/scala/org/apache/zeppelin/flink/FlinkScalaInterpreter.scala
@@ -44,7 +44,6 @@ import org.apache.flink.table.catalog.hive.HiveCatalog
import org.apache.flink.table.functions.{AggregateFunction, ScalarFunction, TableAggregateFunction, TableFunction}
import org.apache.flink.table.module.hive.HiveModule
import org.apache.flink.yarn.cli.FlinkYarnSessionCli
-import org.apache.zeppelin.conf.ZeppelinConfiguration
import org.apache.zeppelin.dep.DependencyResolver
import org.apache.zeppelin.flink.internal.FlinkShell
import org.apache.zeppelin.flink.internal.FlinkShell._
@@ -66,8 +65,7 @@ import scala.tools.nsc.interpreter.{Completion, IMain, IR, JPrintWriter, Results
* @param properties
*/
abstract class FlinkScalaInterpreter(val properties: Properties,
- val flinkScalaClassLoader: ClassLoader,
- val zConf: ZeppelinConfiguration) {
+ val flinkScalaClassLoader: ClassLoader) {
private lazy val LOGGER: Logger = LoggerFactory.getLogger(getClass)
@@ -800,7 +798,12 @@ abstract class FlinkScalaInterpreter(val properties: Properties,
val flinkPackageJars =
if (!StringUtils.isBlank(properties.getProperty("flink.execution.packages", ""))) {
val packages = properties.getProperty("flink.execution.packages")
- val dependencyResolver = new DependencyResolver(System.getProperty("user.home") + "/.m2/repository", zConf)
+ val dependencyResolver = new DependencyResolver(
+ System.getProperty("user.home") + "/.m2/repository",
+ properties.getProperty("zeppelin.proxy.url"),
+ properties.getProperty("zeppelin.proxy.user"),
+ properties.getProperty("zeppelin.proxy.password"),
+ properties.getProperty("zeppelin.interpreter.dep.mvnRepo"))
packages.split(",")
.flatMap(e => JavaConversions.asScalaBuffer(dependencyResolver.load(e)))
.map(e => e.getAbsolutePath).toSeq
diff --git a/markdown/src/main/java/org/apache/zeppelin/markdown/FlexmarkParser.java b/markdown/src/main/java/org/apache/zeppelin/markdown/FlexmarkParser.java
index 82c5b9d354..47fc6b3951 100644
--- a/markdown/src/main/java/org/apache/zeppelin/markdown/FlexmarkParser.java
+++ b/markdown/src/main/java/org/apache/zeppelin/markdown/FlexmarkParser.java
@@ -28,7 +28,6 @@ import com.vladsch.flexmark.util.ast.Node;
import com.vladsch.flexmark.html.HtmlRenderer;
import com.vladsch.flexmark.parser.Parser;
import com.vladsch.flexmark.util.data.MutableDataSet;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,7 +43,7 @@ public class FlexmarkParser implements MarkdownParser {
private Parser parser;
private HtmlRenderer renderer;
- public FlexmarkParser(ZeppelinConfiguration zConf) {
+ public FlexmarkParser(boolean escapeHtml) {
MutableDataSet options = new MutableDataSet();
options.set(Parser.EXTENSIONS, Arrays.asList(StrikethroughExtension.create(),
TablesExtension.create(),
@@ -55,7 +54,7 @@ public class FlexmarkParser implements MarkdownParser {
EmojiExtension.create()));
options.set(HtmlRenderer.SOFT_BREAK, "<br />\n");
options.set(EmojiExtension.USE_IMAGE_TYPE, UNICODE_ONLY);
- options.set(HtmlRenderer.ESCAPE_HTML, zConf.isZeppelinNotebookMarkdownEscapeHtml());
+ options.set(HtmlRenderer.ESCAPE_HTML, escapeHtml);
parser = Parser.builder(options).build();
renderer = HtmlRenderer.builder(options).build();
}
diff --git a/markdown/src/main/java/org/apache/zeppelin/markdown/Markdown.java b/markdown/src/main/java/org/apache/zeppelin/markdown/Markdown.java
index 46733e43a0..8d51fa8140 100644
--- a/markdown/src/main/java/org/apache/zeppelin/markdown/Markdown.java
+++ b/markdown/src/main/java/org/apache/zeppelin/markdown/Markdown.java
@@ -74,7 +74,9 @@ public class Markdown extends Interpreter {
LOGGER.debug("Creating {} markdown interpreter", parserType);
if (MarkdownParserType.FLEXMARK.toString().equals(parserType)) {
- return new FlexmarkParser(zConf);
+ boolean escapeHtml = Boolean.parseBoolean(
+ getProperty("zeppelin.notebook.markdown.escape.html", "true"));
+ return new FlexmarkParser(escapeHtml);
} else {
// default parser
return new Markdown4jParser();
diff --git a/markdown/src/test/java/org/apache/zeppelin/markdown/FlexmarkParserTest.java b/markdown/src/test/java/org/apache/zeppelin/markdown/FlexmarkParserTest.java
index 01540b09c4..ef9327e89e 100644
--- a/markdown/src/test/java/org/apache/zeppelin/markdown/FlexmarkParserTest.java
+++ b/markdown/src/test/java/org/apache/zeppelin/markdown/FlexmarkParserTest.java
@@ -17,7 +17,6 @@
package org.apache.zeppelin.markdown;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.InterpreterResult;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -44,7 +43,6 @@ class FlexmarkParserTest {
Properties props = new Properties();
props.put(Markdown.MARKDOWN_PARSER_TYPE, Markdown.PARSER_TYPE_FLEXMARK);
md = new Markdown(props);
- md.setZeppelinConfiguration(ZeppelinConfiguration.load());
md.open();
}
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/AbstractDependencyResolver.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/AbstractDependencyResolver.java
index b44e096e4a..1b3693a20e 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/AbstractDependencyResolver.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/AbstractDependencyResolver.java
@@ -24,7 +24,6 @@ import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.eclipse.aether.RepositorySystem;
import org.eclipse.aether.RepositorySystemSession;
import org.eclipse.aether.repository.Authentication;
@@ -49,26 +48,28 @@ public abstract class AbstractDependencyResolver {
protected RepositorySystemSession session;
private Proxy proxy = null;
- protected AbstractDependencyResolver(String localRepoPath, ZeppelinConfiguration zConf) {
- if (zConf.getZeppelinProxyUrl() != null) {
+ protected AbstractDependencyResolver(String localRepoPath,
+ String proxyUrl, String proxyUser, String proxyPassword, String mvnRepoUrl) {
+ if (proxyUrl != null) {
try {
- URL proxyUrl = new URL(zConf.getZeppelinProxyUrl());
- Authentication auth = new AuthenticationBuilder().addUsername(zConf.getZeppelinProxyUser()).addPassword(zConf.getZeppelinProxyPassword()).build();
- proxy = new Proxy(proxyUrl.getProtocol(), proxyUrl.getHost(), proxyUrl.getPort(), auth);
+ URL parsedProxyUrl = new URL(proxyUrl);
+ Authentication auth = new AuthenticationBuilder()
+ .addUsername(proxyUser).addPassword(proxyPassword).build();
+ proxy = new Proxy(parsedProxyUrl.getProtocol(), parsedProxyUrl.getHost(),
+ parsedProxyUrl.getPort(), auth);
} catch (MalformedURLException e) {
- LOGGER.error("Proxy Url {} is not valid - skipping Proxy config", zConf.getZeppelinProxyUrl(), e);
+ LOGGER.error("Proxy Url {} is not valid - skipping Proxy config", proxyUrl, e);
}
}
session = Booter.newRepositorySystemSession(system, localRepoPath);
- repos.addAll(Booter.newCentralRepositorys(proxy, zConf)); // add maven central
+ repos.addAll(Booter.newCentralRepositorys(proxy, mvnRepoUrl)); // add maven central
repos.add(Booter.newLocalRepository());
}
- protected AbstractDependencyResolver(String localRepoPath, Proxy proxy,
- ZeppelinConfiguration zConf) {
+ protected AbstractDependencyResolver(String localRepoPath, Proxy proxy, String mvnRepoUrl) {
this.proxy = proxy;
session = Booter.newRepositorySystemSession(system, localRepoPath);
- repos.addAll(Booter.newCentralRepositorys(proxy, zConf)); // add maven central
+ repos.addAll(Booter.newCentralRepositorys(proxy, mvnRepoUrl)); // add maven central
repos.add(Booter.newLocalRepository());
}
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/Booter.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/Booter.java
index b0e89b96a4..16822c6ae7 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/Booter.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/Booter.java
@@ -18,7 +18,6 @@
package org.apache.zeppelin.dep;
import org.apache.commons.lang3.Validate;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.maven.repository.internal.MavenRepositorySystemUtils;
import org.eclipse.aether.DefaultRepositorySystemSession;
import org.eclipse.aether.RepositorySystem;
@@ -83,10 +82,10 @@ public class Booter {
}
public static List<RemoteRepository> newCentralRepositorys(Proxy proxy,
- ZeppelinConfiguration zConf) {
+ String mvnRepoUrl) {
String mvnRepoEnv = System.getenv("ZEPPELIN_INTERPRETER_DEP_MVNREPO");
if (mvnRepoEnv == null) {
- mvnRepoEnv = zConf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_DEP_MVNREPO);
+ mvnRepoEnv = mvnRepoUrl;
}
if (mvnRepoEnv == null) {
mvnRepoEnv = "https://repo1.maven.org/maven2/";
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/DependencyResolver.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/DependencyResolver.java
index cd7eea5a05..e513ac2b66 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/DependencyResolver.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/DependencyResolver.java
@@ -27,7 +27,6 @@ import java.util.List;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.eclipse.aether.RepositoryException;
import org.eclipse.aether.artifact.Artifact;
import org.eclipse.aether.artifact.DefaultArtifact;
@@ -56,12 +55,13 @@ public class DependencyResolver extends AbstractDependencyResolver {
"org.apache.zeppelin:zeppelin-interpreter",
"org.apache.zeppelin:zeppelin-server"};
- public DependencyResolver(String localRepoPath, ZeppelinConfiguration zConf) {
- super(localRepoPath, zConf);
+ public DependencyResolver(String localRepoPath,
+ String proxyUrl, String proxyUser, String proxyPassword, String mvnRepoUrl) {
+ super(localRepoPath, proxyUrl, proxyUser, proxyPassword, mvnRepoUrl);
}
- public DependencyResolver(String localRepoPath, Proxy proxy, ZeppelinConfiguration zConf) {
- super(localRepoPath, proxy, zConf);
+ public DependencyResolver(String localRepoPath, Proxy proxy, String mvnRepoUrl) {
+ super(localRepoPath, proxy, mvnRepoUrl);
}
public List<File> load(String artifact)
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/Interpreter.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/Interpreter.java
index 9c3c9e59f5..bf09cb7106 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/Interpreter.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/Interpreter.java
@@ -22,7 +22,6 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.zeppelin.annotation.Experimental;
import org.apache.zeppelin.annotation.ZeppelinApi;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
import org.apache.zeppelin.scheduler.Scheduler;
import org.apache.zeppelin.scheduler.SchedulerFactory;
@@ -150,7 +149,6 @@ public abstract class Interpreter {
private URL[] classloaderUrls;
protected Properties properties;
protected String userName;
- protected ZeppelinConfiguration zConf;
@ZeppelinApi
public Interpreter(Properties properties) {
@@ -200,10 +198,6 @@ public abstract class Interpreter {
return this.userName;
}
- public void setZeppelinConfiguration(ZeppelinConfiguration zConf) {
- this.zConf = zConf;
- }
-
public void setInterpreterGroup(InterpreterGroup interpreterGroup) {
this.interpreterGroup = interpreterGroup;
}
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterOption.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterOption.java
index f7b17e362e..116d260488 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterOption.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterOption.java
@@ -20,8 +20,6 @@ package org.apache.zeppelin.interpreter;
import java.util.ArrayList;
import java.util.List;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
-
/**
*
*/
@@ -29,7 +27,7 @@ public class InterpreterOption {
public static final transient String SHARED = "shared";
public static final transient String SCOPED = "scoped";
public static final transient String ISOLATED = "isolated";
- private transient ZeppelinConfiguration zConf;
+ private transient boolean usernameForceLowerCase;
// always set it as true, keep this field just for backward compatibility
boolean remote = true;
@@ -60,8 +58,8 @@ public class InterpreterOption {
this.host = host;
}
- public void setConf(ZeppelinConfiguration zConf) {
- this.zConf = zConf;
+ public void setUsernameForceLowerCase(boolean usernameForceLowerCase) {
+ this.usernameForceLowerCase = usernameForceLowerCase;
}
public boolean permissionIsSet() {
@@ -73,7 +71,7 @@ public class InterpreterOption {
}
public List<String> getOwners() {
- if (null != owners && zConf.isUsernameForceLowerCase()) {
+ if (null != owners && usernameForceLowerCase) {
List<String> lowerCaseUsers = new ArrayList<>();
for (String owner : owners) {
lowerCaseUsers.add(owner.toLowerCase());
@@ -117,7 +115,7 @@ public class InterpreterOption {
option.setPermission = other.setPermission;
option.owners = (null == other.owners) ?
new ArrayList<>() : new ArrayList<>(other.owners);
- option.zConf = other.zConf;
+ option.usernameForceLowerCase = other.usernameForceLowerCase;
return option;
}
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/LazyOpenInterpreter.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/LazyOpenInterpreter.java
index ec236ce1f2..f794a58283 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/LazyOpenInterpreter.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/LazyOpenInterpreter.java
@@ -21,7 +21,6 @@ import java.net.URL;
import java.util.List;
import java.util.Properties;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
import org.apache.zeppelin.scheduler.Scheduler;
@@ -205,11 +204,6 @@ public class LazyOpenInterpreter
this.intp.setUserName(userName);
}
- @Override
- public void setZeppelinConfiguration(ZeppelinConfiguration zConf) {
- this.intp.setZeppelinConfiguration(zConf);
- }
-
@Override
public String getUserName() {
return this.intp.getUserName();
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/LifecycleManager.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/LifecycleManager.java
index 8f44b2f962..71528adf6d 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/LifecycleManager.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/LifecycleManager.java
@@ -19,19 +19,20 @@
package org.apache.zeppelin.interpreter;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer;
+import java.util.Properties;
+
/**
* Abstract base class for managing the lifecycle of interpreters
*/
public abstract class LifecycleManager {
- protected ZeppelinConfiguration zConf;
+ protected Properties properties;
protected RemoteInterpreterServer remoteInterpreterServer;
- public LifecycleManager(ZeppelinConfiguration zConf, RemoteInterpreterServer remoteInterpreterServer) {
- this.zConf = zConf;
+ public LifecycleManager(Properties properties, RemoteInterpreterServer remoteInterpreterServer) {
+ this.properties = properties;
this.remoteInterpreterServer = remoteInterpreterServer;
}
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/launcher/InterpreterLauncher.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/launcher/InterpreterLauncher.java
index 7190bea489..79accd19b4 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/launcher/InterpreterLauncher.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/launcher/InterpreterLauncher.java
@@ -20,26 +20,33 @@ package org.apache.zeppelin.interpreter.launcher;
import java.io.IOException;
import java.util.Properties;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.recovery.RecoveryStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECTION_POOL_SIZE;
-
/**
* Component to Launch interpreter process.
*/
public abstract class InterpreterLauncher {
private static final Logger LOGGER = LoggerFactory.getLogger(InterpreterLauncher.class);
- private static final String SPECIAL_CHARACTER="{}()<>&*β|=?;[]$β#~!.\"%/\\:+,`";
+ private static final String SPECIAL_CHARACTER="{}()<>&*'|=?;[]$β#~!.\"%/\\:+,`";
+
+ private static final String CONNECT_TIMEOUT_KEY = "zeppelin.interpreter.connect.timeout";
+ private static final long DEFAULT_CONNECT_TIMEOUT = 600000L;
+
+ private static final String CONNECTION_POOL_SIZE_KEY = "zeppelin.interpreter.connection.poolsize";
+ private static final int DEFAULT_CONNECTION_POOL_SIZE = 100;
- protected final ZeppelinConfiguration zConf;
+ private static final String RECOVERY_ENABLED_KEY = "zeppelin.recovery.storage.class";
+ private static final String NULL_RECOVERY_CLASS =
+ "org.apache.zeppelin.interpreter.recovery.NullRecoveryStorage";
+
+ protected final Properties zProperties;
protected final RecoveryStorage recoveryStorage;
- protected InterpreterLauncher(ZeppelinConfiguration zConf, RecoveryStorage recoveryStorage) {
- this.zConf = zConf;
+ protected InterpreterLauncher(Properties zProperties, RecoveryStorage recoveryStorage) {
+ this.zProperties = zProperties;
this.recoveryStorage = recoveryStorage;
}
@@ -49,21 +56,20 @@ public abstract class InterpreterLauncher {
* @return
*/
protected int getConnectTimeout(InterpreterLaunchContext context) {
- int connectTimeout =
- (int) zConf.getTime(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT);
+ int connectTimeout = (int) Long.parseLong(
+ zProperties.getProperty(CONNECT_TIMEOUT_KEY,
+ String.valueOf(DEFAULT_CONNECT_TIMEOUT)));
Properties properties = context.getProperties();
- if (properties != null && properties.containsKey(
- ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT.getVarName())) {
- connectTimeout = Integer.parseInt(properties.getProperty(
- ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT.getVarName()));
+ if (properties != null && properties.containsKey(CONNECT_TIMEOUT_KEY)) {
+ connectTimeout = Integer.parseInt(properties.getProperty(CONNECT_TIMEOUT_KEY));
}
return connectTimeout;
}
protected int getConnectPoolSize(InterpreterLaunchContext context) {
return Integer.parseInt(context.getProperties().getProperty(
- ZEPPELIN_INTERPRETER_CONNECTION_POOL_SIZE.getVarName(),
- ZEPPELIN_INTERPRETER_CONNECTION_POOL_SIZE.getIntValue() + ""));
+ CONNECTION_POOL_SIZE_KEY,
+ String.valueOf(DEFAULT_CONNECTION_POOL_SIZE)));
}
public static String escapeSpecialCharacter(String command) {
@@ -86,7 +92,7 @@ public abstract class InterpreterLauncher {
*/
public InterpreterClient launch(InterpreterLaunchContext context) throws IOException {
// try to recover it first
- if (zConf.isRecoveryEnabled()) {
+ if (isRecoveryEnabled()) {
InterpreterClient recoveredClient =
recoveryStorage.getInterpreterClient(context.getInterpreterGroupId());
if (recoveredClient != null) {
@@ -106,6 +112,11 @@ public abstract class InterpreterLauncher {
return launchDirectly(context);
}
+ private boolean isRecoveryEnabled() {
+ String recoveryClass = zProperties.getProperty(RECOVERY_ENABLED_KEY, NULL_RECOVERY_CLASS);
+ return !NULL_RECOVERY_CLASS.equals(recoveryClass);
+ }
+
/**
* launch interpreter process directly without recovering.
*
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/lifecycle/NullLifecycleManager.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/lifecycle/NullLifecycleManager.java
index da09c0fbd6..3360982687 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/lifecycle/NullLifecycleManager.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/lifecycle/NullLifecycleManager.java
@@ -18,18 +18,19 @@
package org.apache.zeppelin.interpreter.lifecycle;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.LifecycleManager;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer;
+import java.util.Properties;
+
/**
* Do nothing for the lifecycle of interpreter. User need to explicitly start/stop interpreter.
*/
public class NullLifecycleManager extends LifecycleManager {
- public NullLifecycleManager(ZeppelinConfiguration zConf,
+ public NullLifecycleManager(Properties properties,
RemoteInterpreterServer remoteInterpreterServer) {
- super(zConf, remoteInterpreterServer);
+ super(properties, remoteInterpreterServer);
}
@Override
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/lifecycle/TimeoutLifecycleManager.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/lifecycle/TimeoutLifecycleManager.java
index e6bb97c7b0..a8b163813d 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/lifecycle/TimeoutLifecycleManager.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/lifecycle/TimeoutLifecycleManager.java
@@ -18,13 +18,13 @@
package org.apache.zeppelin.interpreter.lifecycle;
import org.apache.thrift.TException;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.LifecycleManager;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer;
import org.apache.zeppelin.scheduler.ExecutorFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Properties;
import java.util.concurrent.ScheduledExecutorService;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
@@ -41,15 +41,20 @@ public class TimeoutLifecycleManager extends LifecycleManager {
private static final Logger LOGGER = LoggerFactory.getLogger(TimeoutLifecycleManager.class);
+ private static final long DEFAULT_CHECK_INTERVAL = 60000L;
+ private static final long DEFAULT_TIMEOUT_THRESHOLD = 3600000L;
+
private long lastBusyTimeInMillis;
- public TimeoutLifecycleManager(ZeppelinConfiguration zConf,
+ public TimeoutLifecycleManager(Properties properties,
RemoteInterpreterServer remoteInterpreterServer) {
- super(zConf, remoteInterpreterServer);
- long checkInterval = zConf.getTime(ZeppelinConfiguration.ConfVars
- .ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_CHECK_INTERVAL);
- long timeoutThreshold = zConf.getTime(
- ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD);
+ super(properties, remoteInterpreterServer);
+ long checkInterval = Long.parseLong(properties.getProperty(
+ "zeppelin.interpreter.lifecyclemanager.timeout.checkinterval",
+ String.valueOf(DEFAULT_CHECK_INTERVAL)));
+ long timeoutThreshold = Long.parseLong(properties.getProperty(
+ "zeppelin.interpreter.lifecyclemanager.timeout.threshold",
+ String.valueOf(DEFAULT_TIMEOUT_THRESHOLD)));
ScheduledExecutorService checkScheduler = ExecutorFactory.singleton()
.createOrGetScheduled("TimeoutLifecycleManager", 1);
checkScheduler.scheduleAtFixedRate(() -> {
@@ -64,7 +69,7 @@ public class TimeoutLifecycleManager extends LifecycleManager {
LOGGER.debug("Check idle time of interpreter");
}
}, checkInterval, checkInterval, MILLISECONDS);
- LOGGER.info("TimeoutLifecycleManager is started with checkInterval: {}, timeoutThreshold: ΒΈ{}", checkInterval,
+ LOGGER.info("TimeoutLifecycleManager is started with checkInterval: {}, timeoutThreshold: {}", checkInterval,
timeoutThreshold);
}
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/recovery/RecoveryStorage.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/recovery/RecoveryStorage.java
index b7c87c0753..7ca2618c67 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/recovery/RecoveryStorage.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/recovery/RecoveryStorage.java
@@ -17,7 +17,6 @@
package org.apache.zeppelin.interpreter.recovery;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.launcher.InterpreterClient;
import java.io.IOException;
@@ -32,13 +31,9 @@ import java.util.Map;
*/
public abstract class RecoveryStorage {
- protected ZeppelinConfiguration zConf;
protected Map<String, InterpreterClient> restoredClients;
- // TODO(zjffdu) The constructor is inconsistent between base class and its implementation.
- // The implementation actually use InterpreterSettingManager, the interface should also use it.
- public RecoveryStorage(ZeppelinConfiguration zConf) {
- this.zConf = zConf;
+ public RecoveryStorage() {
}
/**
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java
index 6728e7f441..e733fd57f8 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java
@@ -25,7 +25,6 @@ import org.apache.thrift.TException;
import org.apache.thrift.server.TThreadPoolServer;
import org.apache.thrift.transport.TServerSocket;
import org.apache.thrift.transport.TTransportException;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.dep.DependencyResolver;
import org.apache.zeppelin.display.AngularObject;
import org.apache.zeppelin.display.AngularObjectRegistry;
@@ -144,7 +143,7 @@ public class RemoteInterpreterServer extends Thread
// pod (RemoteInterpreterServer)
private boolean isForceShutdown = true;
- private ZeppelinConfiguration zConf;
+ private Properties zProperties;
private static Thread shutdownThread;
@@ -200,10 +199,8 @@ public class RemoteInterpreterServer extends Thread
@Override
public void init(Map<String, String> properties) throws InterpreterRPCException, TException {
- this.zConf = ZeppelinConfiguration.load();
- for (Map.Entry<String, String> entry : properties.entrySet()) {
- this.zConf.setProperty(entry.getKey(), entry.getValue());
- }
+ this.zProperties = new Properties();
+ this.zProperties.putAll(properties);
try {
lifecycleManager = createLifecycleManager();
@@ -213,8 +210,8 @@ public class RemoteInterpreterServer extends Thread
}
if (!isTest) {
- int connectionPoolSize =
- this.zConf.getInt(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECTION_POOL_SIZE);
+ int connectionPoolSize = Integer.parseInt(
+ zProperties.getProperty("zeppelin.interpreter.connection.poolsize", "100"));
LOGGER.info("Creating RemoteInterpreterEventClient with connection pool size: {}",
connectionPoolSize);
intpEventClient = new RemoteInterpreterEventClient(intpEventServerHost, intpEventServerPort,
@@ -249,8 +246,8 @@ public class RemoteInterpreterServer extends Thread
shutDownThread.start();
}
- public ZeppelinConfiguration getConf() {
- return this.zConf;
+ public Properties getProperties() {
+ return this.zProperties;
}
public LifecycleManager getLifecycleManager() {
@@ -270,11 +267,13 @@ public class RemoteInterpreterServer extends Thread
}
private LifecycleManager createLifecycleManager() throws Exception {
- String lifecycleManagerClass = zConf.getLifecycleManagerClass();
+ String lifecycleManagerClass = zProperties.getProperty(
+ "zeppelin.interpreter.lifecyclemanager.class",
+ "org.apache.zeppelin.interpreter.lifecycle.NullLifecycleManager");
Class<?> clazz = Class.forName(lifecycleManagerClass);
LOGGER.info("Creating interpreter lifecycle manager: {}", lifecycleManagerClass);
- return (LifecycleManager) clazz.getConstructor(ZeppelinConfiguration.class, RemoteInterpreterServer.class)
- .newInstance(zConf, this);
+ return (LifecycleManager) clazz.getConstructor(Properties.class, RemoteInterpreterServer.class)
+ .newInstance(zProperties, this);
}
public static void main(String[] args) throws Exception {
@@ -335,7 +334,11 @@ public class RemoteInterpreterServer extends Thread
properties.get("zeppelin.interpreter.output.limit"));
}
- depLoader = new DependencyResolver(localRepoPath, zConf);
+ depLoader = new DependencyResolver(localRepoPath,
+ zProperties.getProperty("zeppelin.proxy.url"),
+ zProperties.getProperty("zeppelin.proxy.user"),
+ zProperties.getProperty("zeppelin.proxy.password"),
+ zProperties.getProperty("zeppelin.interpreter.dep.mvnRepo"));
appLoader = new ApplicationLoader(resourcePool, depLoader);
resultCacheInSeconds =
@@ -361,7 +364,6 @@ public class RemoteInterpreterServer extends Thread
interpreter.setInterpreterGroup(interpreterGroup);
interpreter.setUserName(userName);
- interpreter.setZeppelinConfiguration(zConf);
interpreterGroup.addInterpreterToSession(new LazyOpenInterpreter(interpreter), sessionId);
@@ -484,7 +486,7 @@ public class RemoteInterpreterServer extends Thread
this.intpEventServerHost = host;
this.intpEventServerPort = port;
intpEventClient = new RemoteInterpreterEventClient(intpEventServerHost, intpEventServerPort,
- this.zConf.getInt(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECTION_POOL_SIZE));
+ Integer.parseInt(zProperties.getProperty("zeppelin.interpreter.connection.poolsize", "100")));
intpEventClient.setIntpGroupId(interpreterGroupId);
this.angularObjectRegistry = new AngularObjectRegistry(interpreterGroup.getId(), intpEventClient);
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/SchedulerFactory.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/SchedulerFactory.java
index e3213467fc..a4178704b5 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/SchedulerFactory.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/SchedulerFactory.java
@@ -17,7 +17,6 @@
package org.apache.zeppelin.scheduler;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.util.ExecutorUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,12 +49,23 @@ public class SchedulerFactory {
}
private SchedulerFactory() {
- int threadPoolSize = ZeppelinConfiguration
- .getStaticInt(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_SCHEDULER_POOL_SIZE);
+ int threadPoolSize = getSchedulerPoolSize();
LOGGER.info("Scheduler Thread Pool Size: {}", threadPoolSize);
executor = ExecutorFactory.singleton().createOrGet(SCHEDULER_EXECUTOR_NAME, threadPoolSize);
}
+ private static int getSchedulerPoolSize() {
+ String envValue = System.getenv("ZEPPELIN_INTERPRETER_SCHEDULER_POOL_SIZE");
+ if (envValue != null) {
+ return Integer.parseInt(envValue);
+ }
+ String propValue = System.getProperty("zeppelin.scheduler.threadpool.size");
+ if (propValue != null) {
+ return Integer.parseInt(propValue);
+ }
+ return 100;
+ }
+
public void destroy() {
LOGGER.info("Destroy all executors");
synchronized (schedulers) {
diff --git a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/conf/ZeppelinConfigurationTest.java b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/conf/ZeppelinConfigurationTest.java
deleted file mode 100644
index 55c68d48a2..0000000000
--- a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/conf/ZeppelinConfigurationTest.java
+++ /dev/null
@@ -1,49 +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 static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-
-import java.time.format.DateTimeParseException;
-
-import org.junit.jupiter.api.Test;
-
-class ZeppelinConfigurationTest {
-
- @Test
- void testTimeUnitToMill() {
- assertEquals(10L, ZeppelinConfiguration.timeUnitToMill("10ms"));
- assertEquals(2000L, ZeppelinConfiguration.timeUnitToMill("2s"));
- assertEquals(60000L, ZeppelinConfiguration.timeUnitToMill("1m"));
- assertEquals(3600000L, ZeppelinConfiguration.timeUnitToMill("1h"));
- }
-
- @Test
- void testTimeUnitToMill_WithoutUnit_1() {
- assertThrows(DateTimeParseException.class, () -> {
- ZeppelinConfiguration.timeUnitToMill("60000");
- });
- }
-
- @Test
- void testTimeUnitToMill_WithoutUnit_2() {
- assertThrows(DateTimeParseException.class, () -> {
- ZeppelinConfiguration.timeUnitToMill("0");
- });
- }
-}
diff --git a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/dep/BooterTest.java b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/dep/BooterTest.java
index a2619db308..7ea565d528 100644
--- a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/dep/BooterTest.java
+++ b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/dep/BooterTest.java
@@ -17,7 +17,6 @@
package org.apache.zeppelin.dep;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.eclipse.aether.repository.RemoteRepository;
import org.junit.jupiter.api.Test;
@@ -56,8 +55,8 @@ class BooterTest {
@Test
void getInterpreterMvnRepoPathTest() {
- ZeppelinConfiguration zConf = ZeppelinConfiguration.load("zeppelin-site-test.xml");
- List<RemoteRepository> remoteRepositories = Booter.newCentralRepositorys(null, zConf);
+ String mvnRepoUrl = "https://repo1.maven.org/maven2/,https://repo2.maven.org/maven2/";
+ List<RemoteRepository> remoteRepositories = Booter.newCentralRepositorys(null, mvnRepoUrl);
assertNotNull(remoteRepositories);
assertEquals(2, remoteRepositories.size());
assertEquals("https://repo1.maven.org/maven2/", remoteRepositories.get(0).getUrl());
diff --git a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/dep/DependencyResolverTest.java b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/dep/DependencyResolverTest.java
index f51b683a75..1a41ab4858 100644
--- a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/dep/DependencyResolverTest.java
+++ b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/dep/DependencyResolverTest.java
@@ -18,7 +18,6 @@
package org.apache.zeppelin.dep;
import org.apache.commons.io.FileUtils;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.eclipse.aether.RepositoryException;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
@@ -45,7 +44,7 @@ class DependencyResolverTest {
System.currentTimeMillis());
testPath = tmpDir.getAbsolutePath() + "/test-repo";
testCopyPath = new File(tmpDir, "test-copy-repo");
- resolver = new DependencyResolver(testPath, ZeppelinConfiguration.load());
+ resolver = new DependencyResolver(testPath, null, null, null, null);
}
@AfterAll
diff --git a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/helium/ApplicationLoaderTest.java b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/helium/ApplicationLoaderTest.java
index 3344a38338..39b392eb4a 100644
--- a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/helium/ApplicationLoaderTest.java
+++ b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/helium/ApplicationLoaderTest.java
@@ -18,7 +18,6 @@
package org.apache.zeppelin.helium;
import org.apache.commons.io.FileUtils;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.dep.DependencyResolver;
import org.apache.zeppelin.interpreter.InterpreterOutput;
import org.apache.zeppelin.resource.LocalResourcePool;
@@ -54,7 +53,7 @@ class ApplicationLoaderTest {
// given
LocalResourcePool resourcePool = new LocalResourcePool("pool1");
DependencyResolver dep =
- new DependencyResolver(tmpDir.getAbsolutePath(), ZeppelinConfiguration.load());
+ new DependencyResolver(tmpDir.getAbsolutePath(), null, null, null, null);
ApplicationLoader appLoader = new ApplicationLoader(resourcePool, dep);
HeliumPackage pkg1 = createPackageInfo(MockApplication1.class.getName(), "artifact1");
diff --git a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServerTest.java b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServerTest.java
index c1700cdef1..58ef866a74 100644
--- a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServerTest.java
+++ b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServerTest.java
@@ -82,7 +82,7 @@ class RemoteInterpreterServerTest {
server.getPort()));
server.init(new HashMap<>());
- assertNotNull(server.getConf());
+ assertNotNull(server.getProperties());
assertNotNull(server.getLifecycleManager());
}
diff --git a/zeppelin-plugins/launcher/docker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterLauncher.java b/zeppelin-plugins/launcher/docker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterLauncher.java
index 91cc7a3a5c..c3fb24080a 100644
--- a/zeppelin-plugins/launcher/docker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterLauncher.java
+++ b/zeppelin-plugins/launcher/docker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterLauncher.java
@@ -16,13 +16,13 @@
*/
package org.apache.zeppelin.interpreter.launcher;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.recovery.RecoveryStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Map;
+import java.util.Properties;
/**
* Interpreter Launcher which use shell script to launch the interpreter process.
@@ -32,9 +32,9 @@ public class DockerInterpreterLauncher extends InterpreterLauncher {
private InterpreterLaunchContext context;
- public DockerInterpreterLauncher(ZeppelinConfiguration zConf, RecoveryStorage recoveryStorage)
+ public DockerInterpreterLauncher(Properties zProperties, RecoveryStorage recoveryStorage)
throws IOException {
- super(zConf, recoveryStorage);
+ super(zProperties, recoveryStorage);
}
@Override
@@ -53,17 +53,17 @@ public class DockerInterpreterLauncher extends InterpreterLauncher {
StandardInterpreterLauncher interpreterLauncher = null;
if (isSpark()) {
- interpreterLauncher = new SparkInterpreterLauncher(zConf, recoveryStorage);
+ interpreterLauncher = new SparkInterpreterLauncher(zProperties, recoveryStorage);
} else if (isFlink()) {
- interpreterLauncher = new FlinkInterpreterLauncher(zConf, recoveryStorage);
+ interpreterLauncher = new FlinkInterpreterLauncher(zProperties, recoveryStorage);
} else {
- interpreterLauncher = new StandardInterpreterLauncher(zConf, recoveryStorage);
+ interpreterLauncher = new StandardInterpreterLauncher(zProperties, recoveryStorage);
}
Map<String, String> env = interpreterLauncher.buildEnvFromProperties(context);
return new DockerInterpreterProcess(
- zConf,
- zConf.getDockerContainerImage(),
+ zProperties,
+ zProperties.getProperty("zeppelin.docker.container.image", "apache/zeppelin"),
context.getInterpreterGroupId(),
context.getInterpreterSettingGroup(),
context.getInterpreterSettingName(),
diff --git a/zeppelin-plugins/launcher/docker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcess.java b/zeppelin-plugins/launcher/docker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcess.java
index 643afb2061..f56ec49859 100644
--- a/zeppelin-plugins/launcher/docker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcess.java
+++ b/zeppelin-plugins/launcher/docker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcess.java
@@ -53,8 +53,6 @@ import com.spotify.docker.client.messages.ProgressMessage;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.filefilter.FileFilterUtils;
import org.apache.commons.lang.StringUtils;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
-import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars;
import org.apache.zeppelin.interpreter.launcher.utils.TarFileEntry;
import org.apache.zeppelin.interpreter.launcher.utils.TarUtils;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess;
@@ -63,7 +61,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static java.nio.charset.StandardCharsets.UTF_8;
-import static org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars.ZEPPELIN_SERVER_KERBEROS_KEYTAB;
public class DockerInterpreterProcess extends RemoteInterpreterProcess {
private static final Logger LOGGER = LoggerFactory.getLogger(DockerInterpreterProcess.class);
@@ -93,7 +90,7 @@ public class DockerInterpreterProcess extends RemoteInterpreterProcess {
@VisibleForTesting
boolean uploadLocalLibToContainter = true;
- private ZeppelinConfiguration zConf;
+ private Properties zProperties;
private String zeppelinHome;
private final String containerZeppelinHome;
@@ -107,7 +104,7 @@ public class DockerInterpreterProcess extends RemoteInterpreterProcess {
private static final String CONTAINER_UPLOAD_TAR_DIR = "/tmp/zeppelin-tar";
public DockerInterpreterProcess(
- ZeppelinConfiguration zConf,
+ Properties zProperties,
String containerImage,
String interpreterGroupId,
String interpreterGroupName,
@@ -128,20 +125,22 @@ public class DockerInterpreterProcess extends RemoteInterpreterProcess {
this.properties = properties;
this.envs = new HashMap<>(envs);
- this.zConf = zConf;
+ this.zProperties = zProperties;
this.containerName = interpreterGroupId.toLowerCase();
- containerZeppelinHome = zConf.getString(ConfVars.ZEPPELIN_DOCKER_CONTAINER_HOME);
- containerSparkHome = zConf.getString(ConfVars.ZEPPELIN_DOCKER_CONTAINER_SPARK_HOME);
- uploadLocalLibToContainter = zConf.getBoolean(
- ConfVars.ZEPPELIN_DOCKER_UPLOAD_LOCAL_LIB_TO_CONTAINTER);
+ containerZeppelinHome = zProperties.getProperty(
+ "zeppelin.docker.container.home", "/opt/zeppelin");
+ containerSparkHome = zProperties.getProperty(
+ "zeppelin.docker.container.spark.home", "/opt/spark");
+ uploadLocalLibToContainter = Boolean.parseBoolean(
+ zProperties.getProperty("zeppelin.docker.upload.local.lib.to.container", "true"));
try {
this.zeppelinHome = getZeppelinHome();
} catch (IOException e) {
LOGGER.error(e.getMessage(), e);
}
- dockerHost = zConf.getString(ConfVars.ZEPPELIN_DOCKER_HOST);
+ dockerHost = zProperties.getProperty("zeppelin.docker.host", "http://0.0.0.0:2375");
}
@Override
@@ -328,7 +327,8 @@ public class DockerInterpreterProcess extends RemoteInterpreterProcess {
envs.remove("PATH");
// set container time zone
- envs.put("TZ", zConf.getString(ConfVars.ZEPPELIN_DOCKER_TIME_ZONE));
+ envs.put("TZ", zProperties.getProperty("zeppelin.docker.time.zone",
+ java.util.TimeZone.getDefault().getID()));
List<String> listEnv = new ArrayList<>();
for (Map.Entry<String, String> entry : this.envs.entrySet()) {
@@ -498,7 +498,7 @@ public class DockerInterpreterProcess extends RemoteInterpreterProcess {
copyFiles.putIfAbsent(intpKeytab, intpKeytab);
}
// 3.5) zeppelin server keytab file
- String zeppelinServerKeytab = zConf.getString(ZEPPELIN_SERVER_KERBEROS_KEYTAB);
+ String zeppelinServerKeytab = zProperties.getProperty("zeppelin.server.kerberos.keytab", "");
if (!StringUtils.isBlank(zeppelinServerKeytab)) {
copyFiles.putIfAbsent(zeppelinServerKeytab, zeppelinServerKeytab);
}
@@ -636,9 +636,10 @@ public class DockerInterpreterProcess extends RemoteInterpreterProcess {
private String getZeppelinHome() throws IOException {
// check zeppelinHome is exist
- File fileZeppelinHome = new File(zConf.getZeppelinHome());
+ String zeppelinHomePath = zProperties.getProperty("zeppelin.home", "./");
+ File fileZeppelinHome = new File(zeppelinHomePath);
if (fileZeppelinHome.exists() && fileZeppelinHome.isDirectory()) {
- return zConf.getZeppelinHome();
+ return zeppelinHomePath;
}
throw new IOException("Can't find zeppelin home path!");
diff --git a/zeppelin-plugins/launcher/docker/src/test/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcessTest.java b/zeppelin-plugins/launcher/docker/src/test/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcessTest.java
index 1e9ad7d0bd..d9767ec58e 100644
--- a/zeppelin-plugins/launcher/docker/src/test/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcessTest.java
+++ b/zeppelin-plugins/launcher/docker/src/test/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcessTest.java
@@ -32,7 +32,6 @@ import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.when;
class DockerInterpreterProcessTest {
@@ -40,8 +39,10 @@ class DockerInterpreterProcessTest {
@Test
void testCreateIntpProcess() throws IOException {
+ Properties zProperties = new Properties();
+ zProperties.putAll(zConf.getCompleteConfiguration());
DockerInterpreterLauncher launcher
- = new DockerInterpreterLauncher(zConf, null);
+ = new DockerInterpreterLauncher(zProperties, null);
Properties properties = new Properties();
properties.setProperty(
ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT.getVarName(), "5000");
@@ -62,21 +63,25 @@ class DockerInterpreterProcessTest {
@Test
void testEnv() throws IOException {
- when(zConf.getString(ConfVars.ZEPPELIN_DOCKER_CONTAINER_SPARK_HOME))
- .thenReturn("my-spark-home");
- when(zConf.getBoolean(ConfVars.ZEPPELIN_DOCKER_UPLOAD_LOCAL_LIB_TO_CONTAINTER))
- .thenReturn(false);
- when(zConf.getString(ConfVars.ZEPPELIN_DOCKER_HOST))
- .thenReturn("http://my-docker-host:2375");
-
Properties properties = new Properties();
properties.setProperty(
ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT.getVarName(), "5000");
HashMap<String, String> envs = new HashMap<String, String>();
envs.put("MY_ENV1", "V1");
+ Properties zProps = new Properties();
+ zProps.putAll(zConf.getCompleteConfiguration());
+ zProps.setProperty(
+ ConfVars.ZEPPELIN_DOCKER_CONTAINER_SPARK_HOME.getVarName(),
+ "my-spark-home");
+ zProps.setProperty(
+ ConfVars.ZEPPELIN_DOCKER_UPLOAD_LOCAL_LIB_TO_CONTAINTER.getVarName(),
+ "false");
+ zProps.setProperty(
+ ConfVars.ZEPPELIN_DOCKER_HOST.getVarName(),
+ "http://my-docker-host:2375");
DockerInterpreterProcess intp = spy(new DockerInterpreterProcess(
- zConf,
+ zProps,
"interpreter-container:1.0",
"shared_process",
"sh",
@@ -101,8 +106,10 @@ class DockerInterpreterProcessTest {
HashMap<String, String> envs = new HashMap<String, String>();
envs.put("MY_ENV1", "V1");
+ Properties zProps = new Properties();
+ zProps.putAll(zConf.getCompleteConfiguration());
DockerInterpreterProcess intp = new DockerInterpreterProcess(
- zConf,
+ zProps,
"interpreter-container:1.0",
"shared_process",
"sh",
diff --git a/zeppelin-plugins/launcher/flink/src/main/java/org/apache/zeppelin/interpreter/launcher/FlinkInterpreterLauncher.java b/zeppelin-plugins/launcher/flink/src/main/java/org/apache/zeppelin/interpreter/launcher/FlinkInterpreterLauncher.java
index 847cef2bff..d06bd4b5ed 100644
--- a/zeppelin-plugins/launcher/flink/src/main/java/org/apache/zeppelin/interpreter/launcher/FlinkInterpreterLauncher.java
+++ b/zeppelin-plugins/launcher/flink/src/main/java/org/apache/zeppelin/interpreter/launcher/FlinkInterpreterLauncher.java
@@ -20,7 +20,6 @@ package org.apache.zeppelin.interpreter.launcher;
import com.google.common.base.CharMatcher;
import com.google.common.collect.Sets;
import org.apache.commons.lang3.StringUtils;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.recovery.RecoveryStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -43,8 +42,8 @@ public class FlinkInterpreterLauncher extends StandardInterpreterLauncher {
"local", "remote", "yarn", "yarn-application", "kubernetes-application");
- public FlinkInterpreterLauncher(ZeppelinConfiguration zConf, RecoveryStorage recoveryStorage) {
- super(zConf, recoveryStorage);
+ public FlinkInterpreterLauncher(Properties zProperties, RecoveryStorage recoveryStorage) {
+ super(zProperties, recoveryStorage);
}
@Override
@@ -157,7 +156,7 @@ public class FlinkInterpreterLauncher extends StandardInterpreterLauncher {
}
final String flinkScalaVersion = scalaVersion;
File flinkInterpreterFolder =
- new File(zConf.getInterpreterDir(), "flink");
+ new File(zProperties.getProperty("zeppelin.interpreter.dir", "interpreter"), "flink");
List<File> flinkScalaJars =
Arrays.stream(flinkInterpreterFolder
.listFiles(file -> file.getName().endsWith(".jar")))
diff --git a/zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/K8sStandardInterpreterLauncher.java b/zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/K8sStandardInterpreterLauncher.java
index 7a1ce48275..b08e7848de 100644
--- a/zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/K8sStandardInterpreterLauncher.java
+++ b/zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/K8sStandardInterpreterLauncher.java
@@ -23,8 +23,8 @@ import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
+import java.util.Properties;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.recovery.RecoveryStorage;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterUtils;
import org.slf4j.Logger;
@@ -41,8 +41,8 @@ public class K8sStandardInterpreterLauncher extends InterpreterLauncher {
private static final Logger LOGGER = LoggerFactory.getLogger(K8sStandardInterpreterLauncher.class);
private final KubernetesClient client;
- public K8sStandardInterpreterLauncher(ZeppelinConfiguration zConf, RecoveryStorage recoveryStorage) {
- super(zConf, recoveryStorage);
+ public K8sStandardInterpreterLauncher(Properties zProperties, RecoveryStorage recoveryStorage) {
+ super(zProperties, recoveryStorage);
client = new DefaultKubernetesClient();
}
@@ -68,7 +68,7 @@ public class K8sStandardInterpreterLauncher extends InterpreterLauncher {
//The namespace of zeppelin server can only be read from Config.KUBERNETES_NAMESPACE_PATH while it runs in k8s cluster, it may be different from the namespace of interpreter
String serverNamespace = K8sUtils.getCurrentK8sNamespace();
return String.format("%s.%s.svc",
- zConf.getK8sServiceName(),
+ zProperties.getProperty("zeppelin.k8s.service.name", "zeppelin-server"),
serverNamespace);
} else {
return context.getIntpEventServerHost();
@@ -96,7 +96,7 @@ public class K8sStandardInterpreterLauncher extends InterpreterLauncher {
* Only if a spark interpreter process is running, user impersonation should be possible for --proxy-user
*/
private boolean isUserImpersonateForSparkInterpreter(InterpreterLaunchContext context) {
- return zConf.getZeppelinImpersonateSparkProxyUser() &&
+ return Boolean.parseBoolean(zProperties.getProperty("zeppelin.impersonate.spark.proxy.user", "true")) &&
context.getOption().isUserImpersonate() &&
"spark".equalsIgnoreCase(context.getInterpreterSettingGroup());
}
@@ -107,9 +107,9 @@ public class K8sStandardInterpreterLauncher extends InterpreterLauncher {
return new K8sRemoteInterpreterProcess(
client,
- K8sUtils.getInterpreterNamespace(context.getProperties(), zConf),
- new File(zConf.getK8sTemplatesDir(), "interpreter"),
- zConf.getK8sContainerImage(),
+ K8sUtils.getInterpreterNamespace(context.getProperties(), zProperties),
+ new File(zProperties.getProperty("zeppelin.k8s.template.dir", "k8s"), "interpreter"),
+ zProperties.getProperty("zeppelin.k8s.interpreter.container.image"),
context.getInterpreterGroupId(),
context.getInterpreterSettingGroup(),
context.getInterpreterSettingName(),
@@ -117,12 +117,12 @@ public class K8sStandardInterpreterLauncher extends InterpreterLauncher {
buildEnvFromProperties(context),
getZeppelinService(context),
getZeppelinServiceRpcPort(context),
- zConf.getK8sPortForward(),
- zConf.getK8sSparkContainerImage(),
+ Boolean.parseBoolean(zProperties.getProperty("zeppelin.k8s.portforward", "false")),
+ zProperties.getProperty("zeppelin.k8s.spark.container.image"),
getConnectTimeout(context),
getConnectPoolSize(context),
isUserImpersonateForSparkInterpreter(context),
- zConf.getK8sTimeoutDuringPending());
+ Boolean.parseBoolean(zProperties.getProperty("zeppelin.k8s.timeout.during.pending", "true")));
}
protected Map<String, String> buildEnvFromProperties(InterpreterLaunchContext context) {
diff --git a/zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/K8sUtils.java b/zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/K8sUtils.java
index f064842753..af417e58c6 100644
--- a/zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/K8sUtils.java
+++ b/zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/K8sUtils.java
@@ -29,8 +29,6 @@ import java.util.regex.Pattern;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.StringUtils;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
-
import io.fabric8.kubernetes.client.Config;
public class K8sUtils {
@@ -104,13 +102,13 @@ public class K8sUtils {
/**
* Get the namespace of the interpreter.
- * Check Order: zeppelin.k8s.interpreter.namespace -> getCurrentK8sNamespace() -> zConf.getK8sNamepsace()
+ * Check Order: zeppelin.k8s.interpreter.namespace -> getCurrentK8sNamespace() -> zProperties.getProperty("zeppelin.k8s.namespace")
* @param properties
- * @param zConf
+ * @param zProperties
* @return the interpreter namespace
* @throws IOException
*/
- public static String getInterpreterNamespace(Properties properties, ZeppelinConfiguration zConf) throws IOException {
+ public static String getInterpreterNamespace(Properties properties, Properties zProperties) throws IOException {
if(properties.containsKey("zeppelin.k8s.interpreter.namespace")){
return properties.getProperty("zeppelin.k8s.interpreter.namespace");
}
@@ -118,7 +116,7 @@ public class K8sUtils {
if (isRunningOnKubernetes()) {
return getCurrentK8sNamespace();
} else {
- return zConf.getK8sNamepsace();
+ return zProperties.getProperty("zeppelin.k8s.namespace", "default");
}
}
diff --git a/zeppelin-plugins/launcher/k8s-standard/src/test/java/org/apache/zeppelin/interpreter/launcher/K8sStandardInterpreterLauncherTest.java b/zeppelin-plugins/launcher/k8s-standard/src/test/java/org/apache/zeppelin/interpreter/launcher/K8sStandardInterpreterLauncherTest.java
index 651f4a9e6f..5ec671d9f7 100644
--- a/zeppelin-plugins/launcher/k8s-standard/src/test/java/org/apache/zeppelin/interpreter/launcher/K8sStandardInterpreterLauncherTest.java
+++ b/zeppelin-plugins/launcher/k8s-standard/src/test/java/org/apache/zeppelin/interpreter/launcher/K8sStandardInterpreterLauncherTest.java
@@ -43,7 +43,9 @@ class K8sStandardInterpreterLauncherTest {
void testK8sLauncher() throws IOException {
// given
ZeppelinConfiguration zConf = ZeppelinConfiguration.load();
- K8sStandardInterpreterLauncher launcher = new K8sStandardInterpreterLauncher(zConf, null);
+ Properties zProperties = new Properties();
+ zProperties.putAll(zConf.getCompleteConfiguration());
+ K8sStandardInterpreterLauncher launcher = new K8sStandardInterpreterLauncher(zProperties, null);
Properties properties = new Properties();
properties.setProperty("ENV_1", "VALUE_1");
properties.setProperty("property_1", "value_1");
@@ -72,7 +74,9 @@ class K8sStandardInterpreterLauncherTest {
void testK8sLauncherWithSparkAndUserImpersonate() throws IOException {
// given
ZeppelinConfiguration zConf = ZeppelinConfiguration.load();
- K8sStandardInterpreterLauncher launcher = new K8sStandardInterpreterLauncher(zConf, null);
+ Properties zProperties = new Properties();
+ zProperties.putAll(zConf.getCompleteConfiguration());
+ K8sStandardInterpreterLauncher launcher = new K8sStandardInterpreterLauncher(zProperties, null);
Properties properties = new Properties();
properties.setProperty("ENV_1", "VALUE_1");
properties.setProperty("property_1", "value_1");
@@ -106,7 +110,9 @@ class K8sStandardInterpreterLauncherTest {
void testK8sLauncherWithSparkAndWithoutUserImpersonate() throws IOException {
// given
ZeppelinConfiguration zConf = ZeppelinConfiguration.load();
- K8sStandardInterpreterLauncher launcher = new K8sStandardInterpreterLauncher(zConf, null);
+ Properties zProperties = new Properties();
+ zProperties.putAll(zConf.getCompleteConfiguration());
+ K8sStandardInterpreterLauncher launcher = new K8sStandardInterpreterLauncher(zProperties, null);
Properties properties = new Properties();
properties.setProperty("ENV_1", "VALUE_1");
properties.setProperty("property_1", "value_1");
diff --git a/zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnInterpreterLauncher.java b/zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnInterpreterLauncher.java
index 33ca438d00..d832069aef 100644
--- a/zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnInterpreterLauncher.java
+++ b/zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnInterpreterLauncher.java
@@ -17,7 +17,6 @@
package org.apache.zeppelin.interpreter.launcher;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.recovery.RecoveryStorage;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterUtils;
import org.slf4j.Logger;
@@ -26,6 +25,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
+import java.util.Properties;
/**
* Launcher for running interpreter in yarn container.
@@ -34,8 +34,8 @@ public class YarnInterpreterLauncher extends InterpreterLauncher {
private static final Logger LOGGER = LoggerFactory.getLogger(YarnInterpreterLauncher.class);
- public YarnInterpreterLauncher(ZeppelinConfiguration zConf, RecoveryStorage recoveryStorage) {
- super(zConf, recoveryStorage);
+ public YarnInterpreterLauncher(Properties zProperties, RecoveryStorage recoveryStorage) {
+ super(zProperties, recoveryStorage);
}
@Override
@@ -48,7 +48,7 @@ public class YarnInterpreterLauncher extends InterpreterLauncher {
buildEnvFromProperties(context),
getConnectTimeout(context),
getConnectPoolSize(context),
- zConf);
+ zProperties);
}
protected Map<String, String> buildEnvFromProperties(InterpreterLaunchContext context) {
diff --git a/zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnRemoteInterpreterProcess.java b/zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnRemoteInterpreterProcess.java
index f47eaf68d7..492cf96530 100644
--- a/zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnRemoteInterpreterProcess.java
+++ b/zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnRemoteInterpreterProcess.java
@@ -46,7 +46,6 @@ import org.apache.hadoop.yarn.exceptions.ApplicationNotFoundException;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.Records;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -78,7 +77,7 @@ public class YarnRemoteInterpreterProcess extends RemoteInterpreterProcess {
private String host;
private int port = -1;
- private final ZeppelinConfiguration zConf;
+ private final Properties zProperties;
private final InterpreterLaunchContext launchContext;
private final Properties properties;
private final Map<String, String> envs;
@@ -103,12 +102,12 @@ public class YarnRemoteInterpreterProcess extends RemoteInterpreterProcess {
Map<String, String> envs,
int connectTimeout,
int connectionPoolSize,
- ZeppelinConfiguration zConf) {
+ Properties zProperties) {
super(connectTimeout,
connectionPoolSize,
launchContext.getIntpEventServerHost(),
launchContext.getIntpEventServerPort());
- this.zConf = zConf;
+ this.zProperties = zProperties;
this.launchContext = launchContext;
this.properties = properties;
this.envs = envs;
@@ -326,12 +325,12 @@ public class YarnRemoteInterpreterProcess extends RemoteInterpreterProcess {
vargs.add("-p");
vargs.add(launchContext.getIntpEventServerPort() + "");
vargs.add("-r");
- vargs.add(zConf.getInterpreterPortRange() + "");
+ vargs.add(zProperties.getProperty("zeppelin.interpreter.rpc.portRange", ":") + "");
vargs.add("-i");
vargs.add(launchContext.getInterpreterGroupId());
vargs.add("-l");
vargs.add(ApplicationConstants.Environment.PWD.$() + "/zeppelin/" +
- ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_LOCALREPO.getStringValue()
+ zProperties.getProperty("zeppelin.interpreter.localRepo", "local-repo")
+ "/" + launchContext.getInterpreterSettingName());
vargs.add("-g");
vargs.add(launchContext.getInterpreterSettingName());
@@ -492,7 +491,7 @@ public class YarnRemoteInterpreterProcess extends RemoteInterpreterProcess {
try (ZipOutputStream interpreterZipStream = new ZipOutputStream(new FileOutputStream(interpreterArchive))) {
interpreterZipStream.setLevel(0);
- String zeppelinHomeEnv = zConf.getZeppelinHome();
+ String zeppelinHomeEnv = zProperties.getProperty("zeppelin.home", "./");
if (org.apache.commons.lang3.StringUtils.isBlank(zeppelinHomeEnv)) {
throw new IOException("ZEPPELIN_HOME is not specified");
}
@@ -506,7 +505,7 @@ public class YarnRemoteInterpreterProcess extends RemoteInterpreterProcess {
File interpreterDir = new File(zeppelinHome, "interpreter/" + launchContext.getInterpreterSettingGroup());
addFileToZipStream(interpreterZipStream, interpreterDir, "interpreter");
- File localRepoDir = new File(zConf.getInterpreterLocalRepoPath() + File.separator
+ File localRepoDir = new File(zProperties.getProperty("zeppelin.interpreter.localRepo", "local-repo") + File.separator
+ launchContext.getInterpreterSettingName());
if (localRepoDir.exists() && localRepoDir.isDirectory()) {
LOGGER.debug("Adding localRepoDir {} to interpreter zip: ", localRepoDir.getAbsolutePath());
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/service/InterpreterService.java b/zeppelin-server/src/main/java/org/apache/zeppelin/service/InterpreterService.java
index a845bf869a..d1f330b993 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/service/InterpreterService.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/service/InterpreterService.java
@@ -71,7 +71,13 @@ public class InterpreterService {
String interpreterBaseDir = zConf.getInterpreterDir();
String localRepoPath = zConf.getInterpreterLocalRepoPath();
- final DependencyResolver dependencyResolver = new DependencyResolver(localRepoPath, zConf);
+ final DependencyResolver dependencyResolver = new DependencyResolver(
+ localRepoPath,
+ zConf.getZeppelinProxyUrl(),
+ zConf.getZeppelinProxyUser(),
+ zConf.getZeppelinProxyPassword(),
+ zConf.getString(
+ ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_DEP_MVNREPO));
// TODO(jl): Make a rule between an interpreter name and an installation directory
List<String> possibleInterpreterDirectories = new ArrayList<>();
diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/service/InterpreterServiceTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/service/InterpreterServiceTest.java
index a73c2563d4..d7a89f30ab 100644
--- a/zeppelin-server/src/test/java/org/apache/zeppelin/service/InterpreterServiceTest.java
+++ b/zeppelin-server/src/test/java/org/apache/zeppelin/service/InterpreterServiceTest.java
@@ -102,7 +102,7 @@ class InterpreterServiceTest {
Path specificInterpreterPath =
Files.createDirectory(Paths.get(interpreterDir.toString(), interpreterName));
DependencyResolver dependencyResolver =
- new DependencyResolver(localRepoDir.toString(), ZeppelinConfiguration.load());
+ new DependencyResolver(localRepoDir.toString(), null, null, null, null);
doNothing().when(mockInterpreterSettingManager).refreshInterpreterTemplates();
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
similarity index 100%
rename from zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
rename to zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinLocationStrategy.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinLocationStrategy.java
similarity index 100%
rename from zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinLocationStrategy.java
rename to zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinLocationStrategy.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 2adaa55a42..f26bc54f0e 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
@@ -197,7 +197,7 @@ public class InterpreterSetting {
public Builder setConf(ZeppelinConfiguration zConf) {
interpreterSetting.zConf = zConf;
- interpreterSetting.option.setConf(zConf);
+ interpreterSetting.option.setUsernameForceLowerCase(zConf.isUsernameForceLowerCase());
return this;
}
@@ -664,7 +664,7 @@ public class InterpreterSetting {
public InterpreterSetting setConf(ZeppelinConfiguration zConf) {
this.zConf = zConf;
- this.option.setConf(zConf);
+ this.option.setUsernameForceLowerCase(zConf.isUsernameForceLowerCase());
return this;
}
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 1f25479aa9..e295938220 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
@@ -178,7 +178,10 @@ public class InterpreterSettingManager implements NoteEventListener {
this.interpreterDirPath = Paths.get(zConf.getInterpreterDir());
LOGGER.debug("InterpreterRootPath: {}", interpreterDirPath);
this.dependencyResolver =
- new DependencyResolver(zConf.getString(ConfVars.ZEPPELIN_INTERPRETER_LOCALREPO), zConf);
+ new DependencyResolver(zConf.getString(ConfVars.ZEPPELIN_INTERPRETER_LOCALREPO),
+ zConf.getZeppelinProxyUrl(), zConf.getZeppelinProxyUser(),
+ zConf.getZeppelinProxyPassword(),
+ zConf.getString(ConfVars.ZEPPELIN_INTERPRETER_DEP_MVNREPO));
this.interpreterRepositories = new ArrayList<>();
for (RemoteRepository repo : dependencyResolver.getRepos()) {
this.interpreterRepositories.add(Repository.fromRemoteRepository(repo));
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/install/InstallInterpreter.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/install/InstallInterpreter.java
index 4c9a3ae3cd..3038e01452 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/install/InstallInterpreter.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/install/InstallInterpreter.java
@@ -163,7 +163,8 @@ public class InstallInterpreter {
proxy = new Proxy(proxyUrl.getProtocol(), proxyUrl.getHost(), proxyUrl.getPort(), auth);
}
DependencyResolver depResolver =
- new DependencyResolver(localRepoDir, proxy, zConf);
+ new DependencyResolver(localRepoDir, proxy,
+ zConf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_DEP_MVNREPO));
File installDir = new File(interpreterBaseDir, name);
if (installDir.exists()) {
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
index d131c816e0..1200a64d3f 100644
--- 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
@@ -36,7 +36,6 @@ import java.util.stream.StreamSupport;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.recovery.RecoveryStorage;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterUtils;
import org.slf4j.Logger;
@@ -54,8 +53,8 @@ public class SparkInterpreterLauncher extends StandardInterpreterLauncher {
private static final String DEFAULT_MASTER = "local[*]";
Optional<String> sparkMaster = Optional.empty();
- public SparkInterpreterLauncher(ZeppelinConfiguration zConf, RecoveryStorage recoveryStorage) {
- super(zConf, recoveryStorage);
+ public SparkInterpreterLauncher(Properties zProperties, RecoveryStorage recoveryStorage) {
+ super(zProperties, recoveryStorage);
}
@Override
@@ -111,18 +110,21 @@ public class SparkInterpreterLauncher extends StandardInterpreterLauncher {
// Because `zeppelin.interpreter.forceShutdown` is initialized in RemoteInterpreterServer
// before SparkInterpreter is created.
context.getProperties().put("zeppelin.interpreter.forceShutdown", "false");
- } else if (zConf.isOnlyYarnCluster()){
+ } else if (Boolean.parseBoolean(
+ zProperties.getProperty("zeppelin.spark.only_yarn_cluster", "false"))){
throw new IOException("Only yarn-cluster mode is allowed, please set " +
- ZeppelinConfiguration.ConfVars.ZEPPELIN_SPARK_ONLY_YARN_CLUSTER.getVarName() +
+ "zeppelin.spark.only_yarn_cluster" +
" to false if you want to use other modes.");
}
if (isYarnMode(context) && getDeployMode(context).equals("cluster")) {
if (sparkProperties.containsKey("spark.files")) {
+ String confDir = zProperties.getProperty("zeppelin.conf.dir", "conf");
sparkProperties.put("spark.files", sparkProperties.getProperty("spark.files") + "," +
- zConf.getConfDir() + "/log4j_yarn_cluster.properties");
+ confDir + "/log4j_yarn_cluster.properties");
} else {
- sparkProperties.put("spark.files", zConf.getConfDir() + "/log4j_yarn_cluster.properties");
+ String confDir = zProperties.getProperty("zeppelin.conf.dir", "conf");
+ sparkProperties.put("spark.files", confDir + "/log4j_yarn_cluster.properties");
}
sparkProperties.put("spark.yarn.maxAppAttempts", "1");
}
@@ -144,7 +146,8 @@ public class SparkInterpreterLauncher extends StandardInterpreterLauncher {
try {
List<String> additionalJars = new ArrayList<>();
Path localRepoPath =
- Paths.get(zConf.getInterpreterLocalRepoPath(), context.getInterpreterSettingId());
+ Paths.get(zProperties.getProperty("zeppelin.interpreter.localRepo", "local-repo"),
+ context.getInterpreterSettingId());
if (Files.exists(localRepoPath) && Files.isDirectory(localRepoPath)) {
try (DirectoryStream<Path> localRepoStream = Files.newDirectoryStream(localRepoPath, Files::isRegularFile)) {
List<String> localRepoJars = StreamSupport.stream(localRepoStream.spliterator(),
@@ -154,7 +157,8 @@ public class SparkInterpreterLauncher extends StandardInterpreterLauncher {
}
}
- Path scalaFolder = Paths.get(zConf.getZeppelinHome(), "/interpreter/spark/scala-" + scalaVersion);
+ String zeppelinHome = zProperties.getProperty("zeppelin.home", "..");
+ Path scalaFolder = Paths.get(zeppelinHome, "/interpreter/spark/scala-" + scalaVersion);
if (!scalaFolder.toFile().exists()) {
throw new IOException("spark scala folder " + scalaFolder.toFile() + " doesn't exist");
}
@@ -165,7 +169,7 @@ public class SparkInterpreterLauncher extends StandardInterpreterLauncher {
additionalJars.addAll(scalaJars);
}
// add zeppelin-interpreter-shaded
- Path interpreterFolder = Paths.get(zConf.getZeppelinHome(), "/interpreter");
+ Path interpreterFolder = Paths.get(zeppelinHome, "/interpreter");
try (DirectoryStream<Path> interpreterStream = Files.newDirectoryStream(interpreterFolder, Files::isRegularFile)) {
List<String> interpreterJars = StreamSupport.stream(interpreterStream.spliterator(),
false)
@@ -194,7 +198,8 @@ public class SparkInterpreterLauncher extends StandardInterpreterLauncher {
}
StringJoiner sparkConfSJ = new StringJoiner("|");
- if (context.getOption().isUserImpersonate() && zConf.getZeppelinImpersonateSparkProxyUser()) {
+ if (context.getOption().isUserImpersonate() && Boolean.parseBoolean(
+ zProperties.getProperty("zeppelin.impersonate.spark.proxy.user", "true"))) {
sparkConfSJ.add("--proxy-user");
sparkConfSJ.add(context.getUserName());
sparkProperties.remove("spark.yarn.keytab");
@@ -221,9 +226,9 @@ public class SparkInterpreterLauncher extends StandardInterpreterLauncher {
}
String keytab = properties.getProperty("spark.yarn.keytab",
- zConf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_SERVER_KERBEROS_KEYTAB));
+ zProperties.getProperty("zeppelin.server.kerberos.keytab"));
String principal = properties.getProperty("spark.yarn.principal",
- zConf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_SERVER_KERBEROS_PRINCIPAL));
+ zProperties.getProperty("zeppelin.server.kerberos.principal"));
if (!StringUtils.isBlank(keytab) && !StringUtils.isBlank(principal)) {
env.put("ZEPPELIN_SERVER_KERBEROS_KEYTAB", keytab);
@@ -386,7 +391,7 @@ public class SparkInterpreterLauncher extends StandardInterpreterLauncher {
" 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);
+ String zeppelinHome = zProperties.getProperty("zeppelin.home", "..");
sparkRBasePath = new File(zeppelinHome,
"interpreter" + File.separator + "spark" + File.separator + "R");
} else {
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/StandardInterpreterLauncher.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/StandardInterpreterLauncher.java
index 77e6e7bddc..2a80f98163 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/StandardInterpreterLauncher.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/StandardInterpreterLauncher.java
@@ -20,7 +20,6 @@ package org.apache.zeppelin.interpreter.launcher;
import org.apache.commons.exec.environment.EnvironmentUtils;
import org.apache.commons.lang3.StringUtils;
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.InterpreterOption;
import org.apache.zeppelin.interpreter.InterpreterRunner;
import org.apache.zeppelin.interpreter.recovery.RecoveryStorage;
@@ -33,6 +32,7 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.util.Map;
+import java.util.Properties;
/**
* Interpreter Launcher which use shell script to launch the interpreter process.
@@ -41,8 +41,8 @@ public class StandardInterpreterLauncher extends InterpreterLauncher {
private static final Logger LOGGER = LoggerFactory.getLogger(StandardInterpreterLauncher.class);
- public StandardInterpreterLauncher(ZeppelinConfiguration zConf, RecoveryStorage recoveryStorage) {
- super(zConf, recoveryStorage);
+ public StandardInterpreterLauncher(Properties zProperties, RecoveryStorage recoveryStorage) {
+ super(zProperties, recoveryStorage);
}
@Override
@@ -68,14 +68,20 @@ public class StandardInterpreterLauncher extends InterpreterLauncher {
false);
} else {
// create new remote process
- String localRepoPath = zConf.getInterpreterLocalRepoPath() + File.separator
+ String interpreterLocalRepoPath = zProperties.getProperty("zeppelin.interpreter.localRepo",
+ "local-repo");
+ String localRepoPath = interpreterLocalRepoPath + File.separator
+ context.getInterpreterSettingId();
+ String interpreterDir = zProperties.getProperty("zeppelin.interpreter.dir", "interpreter");
+ String interpreterPortRange = zProperties.getProperty("zeppelin.interpreter.portRange", ":");
+ String remoteRunnerPath = zProperties.getProperty("zeppelin.interpreter.remoterunner",
+ System.getProperty("zeppelin.home") + "/bin/interpreter.sh");
return new ExecRemoteInterpreterProcess(
- context.getIntpEventServerPort(), context.getIntpEventServerHost(), zConf.getInterpreterPortRange(),
- zConf.getInterpreterDir() + "/" + groupName, localRepoPath,
+ context.getIntpEventServerPort(), context.getIntpEventServerHost(), interpreterPortRange,
+ interpreterDir + "/" + groupName, localRepoPath,
buildEnvFromProperties(context), connectTimeout, connectionPoolSize, name,
context.getInterpreterGroupId(), option.isUserImpersonate(),
- runner != null ? runner.getPath() : zConf.getInterpreterRemoteRunnerPath());
+ runner != null ? runner.getPath() : remoteRunnerPath);
}
}
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/FileSystemRecoveryStorage.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/FileSystemRecoveryStorage.java
index 26cecb256e..40448f9a92 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/FileSystemRecoveryStorage.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/FileSystemRecoveryStorage.java
@@ -44,6 +44,7 @@ public class FileSystemRecoveryStorage extends RecoveryStorage {
private static final Logger LOGGER = LoggerFactory.getLogger(FileSystemRecoveryStorage.class);
+ private final ZeppelinConfiguration zConf;
private FileSystemStorage fs;
private Path recoveryDir;
private InterpreterSettingManager interpreterSettingManager;
@@ -51,7 +52,8 @@ public class FileSystemRecoveryStorage extends RecoveryStorage {
public FileSystemRecoveryStorage(ZeppelinConfiguration zConf,
InterpreterSettingManager interpreterSettingManager)
throws IOException {
- super(zConf);
+ super();
+ this.zConf = zConf;
this.interpreterSettingManager = interpreterSettingManager;
String recoveryDirProperty = zConf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_RECOVERY_DIR);
this.fs = new FileSystemStorage(zConf, recoveryDirProperty);
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/LocalRecoveryStorage.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/LocalRecoveryStorage.java
index 078722e2a6..de088306f8 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/LocalRecoveryStorage.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/LocalRecoveryStorage.java
@@ -38,17 +38,20 @@ public class LocalRecoveryStorage extends RecoveryStorage {
private static final Logger LOGGER = LoggerFactory.getLogger(LocalRecoveryStorage.class);
+ private final ZeppelinConfiguration zConf;
private InterpreterSettingManager interpreterSettingManager;
private File recoveryDir;
public LocalRecoveryStorage(ZeppelinConfiguration zConf) {
- super(zConf);
+ super();
+ this.zConf = zConf;
}
public LocalRecoveryStorage(ZeppelinConfiguration zConf,
InterpreterSettingManager interpreterSettingManager)
throws IOException {
- super(zConf);
+ super();
+ this.zConf = zConf;
this.recoveryDir = new File(zConf.getRecoveryDir());
LOGGER.info("Using folder {} to store recovery data", recoveryDir);
if (!this.recoveryDir.exists()) {
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/NullRecoveryStorage.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/NullRecoveryStorage.java
index ec6a3b058d..46af727b79 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/NullRecoveryStorage.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/NullRecoveryStorage.java
@@ -35,7 +35,7 @@ public class NullRecoveryStorage extends RecoveryStorage {
public NullRecoveryStorage(ZeppelinConfiguration zConf,
InterpreterSettingManager interpreterSettingManager)
throws IOException {
- super(zConf);
+ super();
}
@Override
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 859e4e2ddc..6bd2e20232 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
@@ -77,7 +77,6 @@ public class RemoteInterpreter extends Interpreter {
this.sessionId = sessionId;
this.className = className;
this.setUserName(userName);
- this.setZeppelinConfiguration(zConf);
}
public boolean isOpened() {
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/plugin/PluginManager.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/plugin/PluginManager.java
index b229d0f0f5..a81b762527 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/plugin/PluginManager.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/plugin/PluginManager.java
@@ -38,6 +38,7 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Properties;
import jakarta.inject.Inject;
@@ -109,13 +110,24 @@ public class PluginManager {
String launcherClassName = "org.apache.zeppelin.interpreter.launcher." + launcherPlugin;
LOGGER.info("Loading Interpreter Launcher Plugin: {}", launcherClassName);
+ Properties zProperties = new Properties();
+ zProperties.putAll(zConf.getCompleteConfiguration());
+ zProperties.setProperty("zeppelin.home", zConf.getZeppelinHome());
+ zProperties.setProperty("zeppelin.conf.dir", zConf.getConfDir());
+ zProperties.setProperty("zeppelin.interpreter.dir",
+ zConf.getInterpreterDir());
+ zProperties.setProperty("zeppelin.interpreter.localRepo",
+ zConf.getInterpreterLocalRepoPath());
+ zProperties.setProperty("zeppelin.interpreter.remoterunner",
+ zConf.getInterpreterRemoteRunnerPath());
+
if (builtinLauncherClassNames.contains(launcherClassName) ||
Boolean.parseBoolean(System.getProperty("zeppelin.isTest", "false"))) {
try {
return (InterpreterLauncher)
(Class.forName(launcherClassName))
- .getConstructor(ZeppelinConfiguration.class, RecoveryStorage.class)
- .newInstance(zConf, recoveryStorage);
+ .getConstructor(Properties.class, RecoveryStorage.class)
+ .newInstance(zProperties, recoveryStorage);
} catch (InstantiationException | IllegalAccessException | ClassNotFoundException
| NoSuchMethodException | InvocationTargetException e) {
throw new IOException("Fail to instantiate InterpreterLauncher from classpath directly:"
@@ -127,8 +139,8 @@ public class PluginManager {
InterpreterLauncher launcher = null;
try {
launcher = (InterpreterLauncher) (Class.forName(launcherClassName, true, pluginClassLoader))
- .getConstructor(ZeppelinConfiguration.class, RecoveryStorage.class)
- .newInstance(zConf, recoveryStorage);
+ .getConstructor(Properties.class, RecoveryStorage.class)
+ .newInstance(zProperties, recoveryStorage);
} catch (InstantiationException | IllegalAccessException | ClassNotFoundException
| NoSuchMethodException | InvocationTargetException e) {
throw new IOException("Fail to instantiate Launcher " + launcherPlugin +
diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/InterpreterSettingTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/InterpreterSettingTest.java
index f7df6715d5..7d4051687b 100644
--- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/InterpreterSettingTest.java
+++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/InterpreterSettingTest.java
@@ -577,7 +577,7 @@ class InterpreterSettingTest extends AbstractInterpreterTest{
.setGroup("group")
.setOption(interpreterOption)
.setIntepreterSettingManager(interpreterSettingManager)
- .setDependencyResolver(new DependencyResolver("/tmp", zConf))
+ .setDependencyResolver(new DependencyResolver("/tmp", null, null, null, null))
.setConf(zConf)
.create();
diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncherTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncherTest.java
index 52ac5a09b5..a3dbd011e2 100644
--- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncherTest.java
+++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncherTest.java
@@ -55,6 +55,7 @@ class SparkInterpreterLauncherTest {
private String sparkHome;
private String zeppelinHome;
private ZeppelinConfiguration zConf;
+ private Properties zProperties;
@BeforeEach
public void setUp() {
@@ -63,12 +64,22 @@ class SparkInterpreterLauncherTest {
zConf.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_HOME.getVarName(),
new File("..").getAbsolutePath());
zeppelinHome = zConf.getZeppelinHome();
+ zProperties = new Properties();
+ zProperties.putAll(zConf.getCompleteConfiguration());
+ zProperties.setProperty("zeppelin.home", zConf.getZeppelinHome());
+ zProperties.setProperty("zeppelin.conf.dir", zConf.getConfDir());
+ zProperties.setProperty("zeppelin.interpreter.dir",
+ zConf.getInterpreterDir());
+ zProperties.setProperty("zeppelin.interpreter.localRepo",
+ zConf.getInterpreterLocalRepoPath());
+ zProperties.setProperty("zeppelin.interpreter.remoterunner",
+ zConf.getInterpreterRemoteRunnerPath());
LOGGER.info("ZEPPELIN_HOME: " + zeppelinHome);
}
@Test
void testConnectTimeOut() throws IOException {
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zProperties, null);
Properties properties = new Properties();
properties.setProperty("SPARK_HOME", sparkHome);
properties.setProperty(
@@ -91,7 +102,7 @@ class SparkInterpreterLauncherTest {
@Test
void testLocalMode() throws IOException {
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zProperties, null);
Properties properties = new Properties();
properties.setProperty("SPARK_HOME", sparkHome);
properties.setProperty("ENV_1", "");
@@ -121,7 +132,7 @@ class SparkInterpreterLauncherTest {
@Test
void testYarnClientMode_1() throws IOException {
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zProperties, null);
Properties properties = new Properties();
properties.setProperty("SPARK_HOME", sparkHome);
properties.setProperty("property_1", "value_1");
@@ -154,7 +165,7 @@ class SparkInterpreterLauncherTest {
@Test
void testYarnClientMode_2() throws IOException {
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zProperties, null);
Properties properties = new Properties();
properties.setProperty("SPARK_HOME", sparkHome);
properties.setProperty("property_1", "value_1");
@@ -189,7 +200,7 @@ class SparkInterpreterLauncherTest {
@Test
void testYarnClusterMode_1() throws IOException {
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zProperties, null);
Properties properties = new Properties();
properties.setProperty("SPARK_HOME", sparkHome);
properties.setProperty("property_1", "value_1");
@@ -231,7 +242,7 @@ class SparkInterpreterLauncherTest {
@Test
void testYarnClusterMode_2() throws IOException {
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zProperties, null);
Properties properties = new Properties();
properties.setProperty("SPARK_HOME", sparkHome);
properties.setProperty("property_1", "value_1");
@@ -282,7 +293,7 @@ class SparkInterpreterLauncherTest {
@Test
void testYarnClusterMode_3() throws IOException {
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zProperties, null);
Properties properties = new Properties();
properties.setProperty("SPARK_HOME", sparkHome);
properties.setProperty("property_1", "value_1");
@@ -335,7 +346,7 @@ class SparkInterpreterLauncherTest {
@Test
void testDetectSparkScalaVersionDirectStreamCapture() throws Exception {
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zProperties, null);
// Use reflection to access private method
Method detectSparkScalaVersionMethod = SparkInterpreterLauncher.class.getDeclaredMethod(
@@ -354,7 +365,7 @@ class SparkInterpreterLauncherTest {
@Test
void testDetectSparkScalaVersionByReplClassWithNonExistentDirectory() throws Exception {
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zProperties, null);
// Use reflection to access private method
Method detectMethod = SparkInterpreterLauncher.class.getDeclaredMethod(
@@ -379,7 +390,7 @@ class SparkInterpreterLauncherTest {
@Test
void testDetectSparkScalaVersionByReplClassWithFileInsteadOfDirectory() throws Exception {
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zProperties, null);
// Use reflection to access private method
Method detectMethod = SparkInterpreterLauncher.class.getDeclaredMethod(
@@ -414,7 +425,7 @@ class SparkInterpreterLauncherTest {
@Test
void testDetectSparkScalaVersionByReplClassWithValidDirectory() throws Exception {
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zProperties, null);
// Use reflection to access private method
Method detectMethod = SparkInterpreterLauncher.class.getDeclaredMethod(
@@ -443,7 +454,7 @@ class SparkInterpreterLauncherTest {
@Test
void testDetectSparkScalaVersionByReplClassWithEmptyDirectory() throws Exception {
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zProperties, null);
// Use reflection to access private method
Method detectMethod = SparkInterpreterLauncher.class.getDeclaredMethod(
@@ -471,7 +482,7 @@ class SparkInterpreterLauncherTest {
@Test
void testDetectSparkScalaVersionByReplClassWithMultipleJars() throws Exception {
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zProperties, null);
// Use reflection to access private method
Method detectMethod = SparkInterpreterLauncher.class.getDeclaredMethod(
@@ -507,7 +518,7 @@ class SparkInterpreterLauncherTest {
@Test
void testDetectSparkScalaVersionByReplClassWithScala213() throws Exception {
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zProperties, null);
// Use reflection to access private method
Method detectMethod = SparkInterpreterLauncher.class.getDeclaredMethod(
@@ -536,7 +547,7 @@ class SparkInterpreterLauncherTest {
@Test
void testDetectSparkScalaVersionByReplClassWithUnsupportedScalaVersion() throws Exception {
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zProperties, null);
// Use reflection to access private method
Method detectMethod = SparkInterpreterLauncher.class.getDeclaredMethod(
diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/launcher/StandardInterpreterLauncherTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/launcher/StandardInterpreterLauncherTest.java
index a9db3ee05a..a95c90af2a 100644
--- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/launcher/StandardInterpreterLauncherTest.java
+++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/launcher/StandardInterpreterLauncherTest.java
@@ -31,16 +31,26 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
class StandardInterpreterLauncherTest {
- private ZeppelinConfiguration zConf;
+ private Properties zProperties;
@BeforeEach
public void setUp() {
- zConf = ZeppelinConfiguration.load();
+ ZeppelinConfiguration zConf = ZeppelinConfiguration.load();
+ zProperties = new Properties();
+ zProperties.putAll(zConf.getCompleteConfiguration());
+ zProperties.setProperty("zeppelin.home", zConf.getZeppelinHome());
+ zProperties.setProperty("zeppelin.conf.dir", zConf.getConfDir());
+ zProperties.setProperty("zeppelin.interpreter.dir",
+ zConf.getInterpreterDir());
+ zProperties.setProperty("zeppelin.interpreter.localRepo",
+ zConf.getInterpreterLocalRepoPath());
+ zProperties.setProperty("zeppelin.interpreter.remoterunner",
+ zConf.getInterpreterRemoteRunnerPath());
}
@Test
void testLauncher() throws IOException {
- StandardInterpreterLauncher launcher = new StandardInterpreterLauncher(zConf, null);
+ StandardInterpreterLauncher launcher = new StandardInterpreterLauncher(zProperties, null);
Properties properties = new Properties();
properties.setProperty("ENV_1", "VALUE_1");
properties.setProperty("property_1", "value_1");
@@ -55,7 +65,6 @@ class StandardInterpreterLauncherTest {
assertEquals(".//local-repo/groupId", interpreterProcess.getLocalRepoDir());
assertEquals(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT.getLongValue(),
interpreterProcess.getConnectTimeout());
- assertEquals(zConf.getInterpreterRemoteRunnerPath(), interpreterProcess.getInterpreterRunner());
assertTrue(interpreterProcess.getEnv().size() >= 2);
assertEquals("VALUE_1", interpreterProcess.getEnv().get("ENV_1"));
assertTrue(interpreterProcess.getEnv().containsKey("INTERPRETER_GROUP_ID"));
@@ -65,7 +74,7 @@ class StandardInterpreterLauncherTest {
@Test
void testConnectTimeOut() throws IOException {
- StandardInterpreterLauncher launcher = new StandardInterpreterLauncher(zConf, null);
+ StandardInterpreterLauncher launcher = new StandardInterpreterLauncher(zProperties, null);
Properties properties = new Properties();
properties.setProperty(
ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT.getVarName(), "10000");
@@ -79,7 +88,6 @@ class StandardInterpreterLauncherTest {
assertEquals(".//interpreter/groupName", interpreterProcess.getInterpreterDir());
assertEquals(".//local-repo/groupId", interpreterProcess.getLocalRepoDir());
assertEquals(10000, interpreterProcess.getConnectTimeout());
- assertEquals(zConf.getInterpreterRemoteRunnerPath(), interpreterProcess.getInterpreterRunner());
assertTrue(interpreterProcess.getEnv().size() >= 1);
assertTrue(interpreterProcess.getEnv().containsKey("INTERPRETER_GROUP_ID"));
assertEquals(true, interpreterProcess.isUserImpersonated());