[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 <noreply@anthropic.com>

Author: jongyoul

flink/flink-scala-2.12/src/main/java/org/apache/zeppelin/flink/FlinkInterpreter.java

@@ -22,7 +22,6 @@
 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 @@ private FlinkScalaInterpreter loadFlinkScalaInterpreter() throws Exception {
     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

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,

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

markdown/src/main/java/org/apache/zeppelin/markdown/FlexmarkParser.java

@@ -28,7 +28,6 @@
 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 FlexmarkParser(ZeppelinConfiguration zConf) {
             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();
   }

markdown/src/main/java/org/apache/zeppelin/markdown/Markdown.java

@@ -74,7 +74,9 @@ public MarkdownParser createMarkdownParser(String parserType) {
     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();

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 @@ public void setUp() throws Exception {
     Properties props = new Properties();
     props.put(Markdown.MARKDOWN_PARSER_TYPE, Markdown.PARSER_TYPE_FLEXMARK);
     md = new Markdown(props);
-    md.setZeppelinConfiguration(ZeppelinConfiguration.load());
     md.open();
   }
 

zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/AbstractDependencyResolver.java

@@ -24,7 +24,6 @@
 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());
   }
 

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 @@ static String resolveLocalRepoPath(String localRepoPath) {
   }
 
   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/";

zeppelin-interpreter/src/main/java/org/apache/zeppelin/dep/DependencyResolver.java

@@ -27,7 +27,6 @@
 
 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)

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/Interpreter.java

@@ -22,7 +22,6 @@
 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 Scheduler getScheduler() {
   private URL[] classloaderUrls;
   protected Properties properties;
   protected String userName;
-  protected ZeppelinConfiguration zConf;
 
   @ZeppelinApi
   public Interpreter(Properties properties) {
@@ -200,10 +198,6 @@ public String getUserName() {
     return this.userName;
   }
 
-  public void setZeppelinConfiguration(ZeppelinConfiguration zConf) {
-    this.zConf = zConf;
-  }
-
   public void setInterpreterGroup(InterpreterGroup interpreterGroup) {
     this.interpreterGroup = interpreterGroup;
   }