Decouple zeppelin-server from Hadoop

zeppelin-plugins/launcher/yarn/pom.xml

@@ -42,12 +42,18 @@
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-client-api</artifactId>
-      <scope>provided</scope>
+      <scope>compile</scope>
     </dependency>
 
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-client-runtime</artifactId>
+      <scope>compile</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-api</artifactId>
       <scope>provided</scope>
     </dependency>
   </dependencies>

zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/YarnAppMonitor.java → zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/YarnAppMonitor.java

@@ -17,6 +17,13 @@
 
 package org.apache.zeppelin.interpreter;
 
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ApplicationReport;
 import org.apache.hadoop.yarn.api.records.YarnApplicationState;
@@ -29,16 +36,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Iterator;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
-/**
- * This class will launch a thread to check yarn app status regularly.
- */
+/** This class launches a thread to check YARN application status regularly. */
 public class YarnAppMonitor {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(YarnAppMonitor.class);
@@ -59,52 +57,64 @@ private YarnAppMonitor() {
     try {
       this.yarnClient = YarnClient.createYarnClient();
       YarnConfiguration yarnConf = new YarnConfiguration();
-      // disable timeline service as we only query yarn app here.
-      // Otherwise we may hit this kind of ERROR:
-      // java.lang.ClassNotFoundException: com.sun.jersey.api.client.config.ClientConfig
+      yarnConf.setClassLoader(YarnAppMonitor.class.getClassLoader());
+      // Disable the timeline service because this client only queries application state.
       yarnConf.set("yarn.timeline-service.enabled", "false");
       yarnClient.init(yarnConf);
       yarnClient.start();
-      this.executor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("YarnAppsMonitor-Thread"));
+      this.executor =
+          Executors.newSingleThreadScheduledExecutor(
+              new NamedThreadFactory("YarnAppsMonitor-Thread"));
       this.apps = new ConcurrentHashMap<>();
-      this.executor.scheduleAtFixedRate(() -> {
-                try {
-                  Iterator<Map.Entry<ApplicationId, RemoteInterpreterManagedProcess>> iter = apps.entrySet().iterator();
-                  while (iter.hasNext()) {
-                    Map.Entry<ApplicationId, RemoteInterpreterManagedProcess> entry = iter.next();
-                    ApplicationId appId = entry.getKey();
-                    RemoteInterpreterManagedProcess interpreterManagedProcess = entry.getValue();
-                    ApplicationReport appReport = yarnClient.getApplicationReport(appId);
-                    if (appReport.getYarnApplicationState() == YarnApplicationState.FAILED ||
-                            appReport.getYarnApplicationState() == YarnApplicationState.KILLED) {
-                      String yarnDiagnostics = appReport.getDiagnostics();
-                      interpreterManagedProcess.processStopped("Yarn diagnostics: " + yarnDiagnostics);
-                      iter.remove();
-                      LOGGER.info("Remove {} from YarnAppMonitor, because its state is {}", appId,
-                              appReport.getYarnApplicationState());
-                    } else if (appReport.getYarnApplicationState() == YarnApplicationState.FINISHED) {
-                      iter.remove();
-                      LOGGER.info("Remove {} from YarnAppMonitor, because its state is {}", appId,
-                              appReport.getYarnApplicationState());
-                    }
-                  }
-                } catch (Exception e) {
-                  LOGGER.warn("Fail to check yarn app status", e);
-                }
-              },
-          ZeppelinConfiguration
-              .getStaticInt(ConfVars.ZEPPELIN_INTERPRETER_YARN_MONITOR_INTERVAL_SECS),
-          ZeppelinConfiguration
-              .getStaticInt(ConfVars.ZEPPELIN_INTERPRETER_YARN_MONITOR_INTERVAL_SECS),
-              TimeUnit.SECONDS);
+      int monitorInterval =
+          ZeppelinConfiguration.getStaticInt(
+              ConfVars.ZEPPELIN_INTERPRETER_YARN_MONITOR_INTERVAL_SECS);
+      this.executor.scheduleAtFixedRate(
+          this::checkApplications,
+          monitorInterval,
+          monitorInterval,
+          TimeUnit.SECONDS);
 
       LOGGER.info("YarnAppMonitor is started");
     } catch (Throwable e) {
       LOGGER.warn("Fail to initialize YarnAppMonitor", e);
     }
   }
 
-  public void addYarnApp(ApplicationId appId, RemoteInterpreterManagedProcess interpreterManagedProcess) {
+  private void checkApplications() {
+    try {
+      Iterator<Map.Entry<ApplicationId, RemoteInterpreterManagedProcess>> iter =
+          apps.entrySet().iterator();
+      while (iter.hasNext()) {
+        Map.Entry<ApplicationId, RemoteInterpreterManagedProcess> entry = iter.next();
+        ApplicationId appId = entry.getKey();
+        RemoteInterpreterManagedProcess interpreterManagedProcess = entry.getValue();
+        ApplicationReport appReport = yarnClient.getApplicationReport(appId);
+        YarnApplicationState applicationState = appReport.getYarnApplicationState();
+        if (applicationState == YarnApplicationState.FAILED
+            || applicationState == YarnApplicationState.KILLED) {
+          interpreterManagedProcess.processStopped(
+              "Yarn diagnostics: " + appReport.getDiagnostics());
+          iter.remove();
+          LOGGER.info(
+              "Remove {} from YarnAppMonitor, because its state is {}",
+              appId,
+              applicationState);
+        } else if (applicationState == YarnApplicationState.FINISHED) {
+          iter.remove();
+          LOGGER.info(
+              "Remove {} from YarnAppMonitor, because its state is {}",
+              appId,
+              applicationState);
+        }
+      }
+    } catch (Exception e) {
+      LOGGER.warn("Fail to check yarn app status", e);
+    }
+  }
+
+  public void addYarnApp(
+      ApplicationId appId, RemoteInterpreterManagedProcess interpreterManagedProcess) {
     LOGGER.info("Add {} to YarnAppMonitor", appId);
     this.apps.put(appId, interpreterManagedProcess);
   }

zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnProcessLaunchObserver.java

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.interpreter.launcher;
+
+import java.util.function.BiConsumer;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.zeppelin.interpreter.YarnAppMonitor;
+import org.apache.zeppelin.interpreter.remote.ProcessLaunchObserver;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterManagedProcess;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Detects an application submitted by a process launcher and registers it for YARN monitoring. */
+public class YarnProcessLaunchObserver implements ProcessLaunchObserver {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(YarnProcessLaunchObserver.class);
+  private static final Pattern YARN_APP_PATTERN = Pattern.compile("Submitted application (\\w+)");
+
+  private final BiConsumer<ApplicationId, RemoteInterpreterManagedProcess> appConsumer;
+
+  public YarnProcessLaunchObserver() {
+    this((appId, process) -> YarnAppMonitor.get().addYarnApp(appId, process));
+  }
+
+  YarnProcessLaunchObserver(
+      BiConsumer<ApplicationId, RemoteInterpreterManagedProcess> appConsumer) {
+    this.appConsumer = appConsumer;
+  }
+
+  @Override
+  public void onProcessLaunch(
+      String launchOutput, RemoteInterpreterManagedProcess interpreterProcess) {
+    Matcher matcher = YARN_APP_PATTERN.matcher(launchOutput);
+    if (matcher.find()) {
+      String appId = matcher.group(1);
+      LOGGER.info("Detected yarn app: {}, add it to YarnAppMonitor", appId);
+      appConsumer.accept(ApplicationId.fromString(appId), interpreterProcess);
+    }
+  }
+}

zeppelin-plugins/launcher/yarn/src/main/java/org/apache/zeppelin/interpreter/launcher/YarnRemoteInterpreterProcess.java

@@ -113,6 +113,7 @@ public YarnRemoteInterpreterProcess(
     this.properties = properties;
     this.envs = envs;
     this.hadoopConf = new YarnConfiguration();
+    this.hadoopConf.setClassLoader(YarnRemoteInterpreterProcess.class.getClassLoader());
     // Add core-site.xml and yarn-site.xml. This is for integration test where using MiniHadoopCluster.
     if (properties.containsKey("HADOOP_CONF_DIR") &&
             !StringUtils.isBlank(properties.getProperty("HADOOP_CONF_DIR"))) {

zeppelin-plugins/launcher/yarn/src/main/resources/META-INF/services/org.apache.zeppelin.interpreter.remote.ProcessLaunchObserver

@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.zeppelin.interpreter.launcher.YarnProcessLaunchObserver

zeppelin-plugins/launcher/yarn/src/test/java/org/apache/zeppelin/interpreter/launcher/YarnProcessLaunchObserverTest.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.interpreter.launcher;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.Mockito.mock;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterManagedProcess;
+import org.junit.jupiter.api.Test;
+
+class YarnProcessLaunchObserverTest {
+
+  @Test
+  void detectsSubmittedYarnApplication() {
+    AtomicReference<ApplicationId> detectedApp = new AtomicReference<>();
+    RemoteInterpreterManagedProcess process = mock(RemoteInterpreterManagedProcess.class);
+    YarnProcessLaunchObserver observer =
+        new YarnProcessLaunchObserver((appId, ignored) -> detectedApp.set(appId));
+
+    observer.onProcessLaunch(
+        "INFO Client: Submitted application application_1720000000000_0042", process);
+
+    assertEquals("application_1720000000000_0042", detectedApp.get().toString());
+  }
+
+  @Test
+  void ignoresLaunchOutputWithoutSubmittedApplication() {
+    AtomicReference<ApplicationId> detectedApp = new AtomicReference<>();
+    YarnProcessLaunchObserver observer =
+        new YarnProcessLaunchObserver((appId, ignored) -> detectedApp.set(appId));
+
+    observer.onProcessLaunch(
+        "INFO Client: Application report for application_1720000000000_0042", null);
+
+    assertNull(detectedApp.get());
+  }
+}

zeppelin-plugins/notebookrepo/filesystem/pom.xml

@@ -30,18 +30,28 @@
 
     <artifactId>notebookrepo-filesystem</artifactId>
     <packaging>jar</packaging>
-    <name>Zeppelin: Plugin FileSystemNotebookRepo</name>
-    <description>NotebookRepo implementation based on Hadoop FileSystem</description>
+    <name>Zeppelin: Plugin Hadoop FileSystem Storage</name>
+    <description>Notebook, configuration, and recovery storage based on Hadoop FileSystem</description>
 
     <properties>
         <adl.sdk.version>2.1.4</adl.sdk.version>
         <plugin.name>NotebookRepo/FileSystemNotebookRepo</plugin.name>
     </properties>
 
     <dependencies>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-client-api</artifactId>
+            <scope>compile</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.hadoop</groupId>
             <artifactId>hadoop-client-runtime</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
             <scope>provided</scope>
         </dependency>
     </dependencies>

zeppelin-server/src/main/java/org/apache/zeppelin/healthcheck/HdfsHealthCheck.java → zeppelin-plugins/notebookrepo/filesystem/src/main/java/org/apache/zeppelin/healthcheck/HdfsHealthCheck.java

@@ -34,7 +34,7 @@ public class HdfsHealthCheck extends HealthCheck {
    */
   public HdfsHealthCheck(FileSystemStorage fs, Path path) {
     this.fs = fs;
-    this.path= path;
+    this.path = path;
   }
   @Override
   protected Result check() throws Exception {

zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/recovery/FileSystemRecoveryStorage.java → zeppelin-plugins/notebookrepo/filesystem/src/main/java/org/apache/zeppelin/interpreter/recovery/FileSystemRecoveryStorage.java

@@ -53,7 +53,8 @@ public FileSystemRecoveryStorage(ZeppelinConfiguration zConf,
       throws IOException {
     super(zConf);
     this.interpreterSettingManager = interpreterSettingManager;
-    String recoveryDirProperty = zConf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_RECOVERY_DIR);
+    String recoveryDirProperty =
+        zConf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_RECOVERY_DIR);
     this.fs = new FileSystemStorage(zConf, recoveryDirProperty);
     LOGGER.info("Creating FileSystem: " + this.fs.getFs().getClass().getName() +
         " for Zeppelin Recovery.");

zeppelin-server/src/main/java/org/apache/zeppelin/notebook/FileSystemStorage.java → zeppelin-plugins/notebookrepo/filesystem/src/main/java/org/apache/zeppelin/notebook/FileSystemStorage.java

@@ -82,6 +82,7 @@ public class FileSystemStorage {
   public FileSystemStorage(ZeppelinConfiguration zConf, String path) throws IOException {
     this.zConf = zConf;
     this.hadoopConf = new Configuration();
+    this.hadoopConf.setClassLoader(FileSystemStorage.class.getClassLoader());
     URI zepConfigURI;
     URI defaultFSURI;
 
@@ -169,7 +170,7 @@ public List<Path> call() throws IOException {
     });
   }
 
-  // recursive search path, (TODO zjffdu, list folder in sub folder on demand, instead of load all
+  // recursive search path, (TODO(zjffdu): list folder in sub folder on demand, instead of load all
   // data when zeppelin server start)
   public List<Path> listAll(final Path path) throws IOException {
     return callHdfsOperation(new HdfsOperation<List<Path>>() {
@@ -213,17 +214,19 @@ public String call() throws IOException {
         LOGGER.debug("Read from file: {}", file);
         ByteArrayOutputStream noteBytes = new ByteArrayOutputStream();
         IOUtils.copyBytes(fs.open(file), noteBytes, hadoopConf);
-        return noteBytes.toString(zConf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_ENCODING));
+        return noteBytes.toString(
+            zConf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_ENCODING));
       }
     });
   }
 
   public void writeFile(final String content, final Path file, boolean writeTempFileFirst)
       throws IOException {
-      writeFile(content, file, writeTempFileFirst, null);
+    writeFile(content, file, writeTempFileFirst, null);
   }
 
-  public void writeFile(final String content, final Path file, boolean writeTempFileFirst, Set<PosixFilePermission> permissions)
+  public void writeFile(final String content, final Path file, boolean writeTempFileFirst,
+                        Set<PosixFilePermission> permissions)
       throws IOException {
     FsPermission fsPermission;
     if (permissions == null || permissions.isEmpty()) {