From 1806fe2e27fba6b9fc5f915a7c809fe54314597d Mon Sep 17 00:00:00 2001 From: HwangRock Date: Sat, 5 Sep 2026 22:13:31 +0900 Subject: [PATCH 1/2] [ZEPPELIN-5745] Add headless CLI to run a note without starting the server Add a bin/run-note.sh CLI that executes a single note headlessly -- no Jetty/REST/WebSocket -- by assembling the interpreter runtime directly, in the spirit of papermill for Jupyter. Supports parameter injection via -p (${var} placeholders) and saving results to a separate note via -o. A failed paragraph exits non-zero so CI/batch callers detect the failure. On exit the process tears down its interpreter processes, RemoteScheduler pools, the event server, and ExecutorFactory pools, then System.exit()s so the JVM does not hang on the runtime's non-daemon threads. --- bin/run-note.sh | 45 ++++ .../cli/HeadlessAngularObjectListener.java | 49 ++++ .../cli/HeadlessApplicationEventListener.java | 59 +++++ .../cli/HeadlessNoteEventListener.java | 70 ++++++ .../notebook/cli/HeadlessProcessListener.java | 183 ++++++++++++++ .../zeppelin/notebook/cli/NotebookRunner.java | 168 +++++++++++++ .../notebook/cli/NotebookRunnerContext.java | 169 +++++++++++++ .../notebook/cli/RunNoteCliOptions.java | 123 ++++++++++ .../notebook/cli/CliTestFixtures.java | 90 +++++++ .../cli/NotebookRunnerContextTest.java | 154 ++++++++++++ .../cli/NotebookRunnerIntegrationTest.java | 225 ++++++++++++++++++ .../cli/NotebookRunnerOutputSaveTest.java | 168 +++++++++++++ .../NotebookRunnerParamSubstitutionTest.java | 137 +++++++++++ .../cli/NotebookRunnerPrototypeTest.java | 89 +++++++ 14 files changed, 1729 insertions(+) create mode 100755 bin/run-note.sh create mode 100644 zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/HeadlessAngularObjectListener.java create mode 100644 zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/HeadlessApplicationEventListener.java create mode 100644 zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/HeadlessNoteEventListener.java create mode 100644 zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/HeadlessProcessListener.java create mode 100644 zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/NotebookRunner.java create mode 100644 zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/NotebookRunnerContext.java create mode 100644 zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/RunNoteCliOptions.java create mode 100644 zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/CliTestFixtures.java create mode 100644 zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerContextTest.java create mode 100644 zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerIntegrationTest.java create mode 100644 zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerOutputSaveTest.java create mode 100644 zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerParamSubstitutionTest.java create mode 100644 zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerPrototypeTest.java diff --git a/bin/run-note.sh b/bin/run-note.sh new file mode 100755 index 00000000000..836e16e043b --- /dev/null +++ b/bin/run-note.sh @@ -0,0 +1,45 @@ +#!/bin/bash +# +# 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. +# +# Run a Zeppelin note headlessly, without starting Zeppelin Server +# + +bin=$(dirname "${BASH_SOURCE-$0}") +bin=$(cd "${bin}">/dev/null; pwd) + +. "${bin}/common.sh" + +ZEPPELIN_RUN_NOTE_MAIN=org.apache.zeppelin.notebook.cli.NotebookRunner +ZEPPELIN_LOGFILE="${ZEPPELIN_LOG_DIR}/run-note.log" +JAVA_OPTS+=" -Dzeppelin.log.file=${ZEPPELIN_LOGFILE}" + +if [[ -d "${ZEPPELIN_HOME}/zeppelin-server/target/classes" ]]; then + ZEPPELIN_CLASSPATH+=":${ZEPPELIN_HOME}/zeppelin-server/target/classes" +fi + +if [[ -d "${ZEPPELIN_HOME}/zeppelin-interpreter/target/classes" ]]; then + ZEPPELIN_CLASSPATH+=":${ZEPPELIN_HOME}/zeppelin-interpreter/target/classes" +fi + +addJarInDir "${ZEPPELIN_HOME}/zeppelin-interpreter/target/lib" +addJarInDir "${ZEPPELIN_HOME}/zeppelin-server/target/lib" +addJarInDir "${ZEPPELIN_HOME}/lib" +addJarInDir "${ZEPPELIN_HOME}/lib/interpreter" + +CLASSPATH+=":${ZEPPELIN_CLASSPATH}" +$ZEPPELIN_RUNNER $JAVA_OPTS -cp $CLASSPATH $ZEPPELIN_RUN_NOTE_MAIN ${@} diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/HeadlessAngularObjectListener.java b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/HeadlessAngularObjectListener.java new file mode 100644 index 00000000000..623374018c4 --- /dev/null +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/HeadlessAngularObjectListener.java @@ -0,0 +1,49 @@ +/* + * 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.notebook.cli; + +import org.apache.zeppelin.display.AngularObject; +import org.apache.zeppelin.display.AngularObjectRegistryListener; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Headless implementation of {@link AngularObjectRegistryListener}. There is no UI to broadcast + * angular object changes to in a headless run, so this only logs at debug level. + */ +public class HeadlessAngularObjectListener implements AngularObjectRegistryListener { + + private static final Logger LOGGER = LoggerFactory.getLogger(HeadlessAngularObjectListener.class); + + @Override + public void onAddAngularObject(String interpreterGroupId, AngularObject angularObject) { + LOGGER.debug("Angular object added in group {}: {}", interpreterGroupId, + angularObject.getName()); + } + + @Override + public void onUpdateAngularObject(String interpreterGroupId, AngularObject angularObject) { + LOGGER.debug("Angular object updated in group {}: {}", interpreterGroupId, + angularObject.getName()); + } + + @Override + public void onRemoveAngularObject(String interpreterGroupId, AngularObject angularObject) { + LOGGER.debug("Angular object removed in group {}: {}", interpreterGroupId, + angularObject.getName()); + } +} diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/HeadlessApplicationEventListener.java b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/HeadlessApplicationEventListener.java new file mode 100644 index 00000000000..c82e107f152 --- /dev/null +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/HeadlessApplicationEventListener.java @@ -0,0 +1,59 @@ +/* + * 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.notebook.cli; + +import org.apache.zeppelin.helium.ApplicationEventListener; +import org.apache.zeppelin.helium.HeliumPackage; +import org.apache.zeppelin.interpreter.InterpreterResult; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Headless implementation of {@link ApplicationEventListener}. Helium applications are out of + * scope for headless note execution (no UI to render them into), so this only logs at debug + * level. + */ +public class HeadlessApplicationEventListener implements ApplicationEventListener { + + private static final Logger LOGGER = + LoggerFactory.getLogger(HeadlessApplicationEventListener.class); + + @Override + public void onOutputAppend(String noteId, String paragraphId, int index, String appId, + String output) { + LOGGER.debug("Helium app {} output append for note {} paragraph {}", appId, noteId, + paragraphId); + } + + @Override + public void onOutputUpdated(String noteId, String paragraphId, int index, String appId, + InterpreterResult.Type type, String output) { + LOGGER.debug("Helium app {} output updated for note {} paragraph {}", appId, noteId, + paragraphId); + } + + @Override + public void onLoad(String noteId, String paragraphId, String appId, HeliumPackage pkg) { + LOGGER.debug("Helium app {} loaded for note {} paragraph {}", appId, noteId, paragraphId); + } + + @Override + public void onStatusChange(String noteId, String paragraphId, String appId, String status) { + LOGGER.debug("Helium app {} status changed to {} for note {} paragraph {}", appId, status, + noteId, paragraphId); + } +} diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/HeadlessNoteEventListener.java b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/HeadlessNoteEventListener.java new file mode 100644 index 00000000000..9f7b4876d27 --- /dev/null +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/HeadlessNoteEventListener.java @@ -0,0 +1,70 @@ +/* + * 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.notebook.cli; + +import org.apache.zeppelin.notebook.Note; +import org.apache.zeppelin.notebook.NoteEventListener; +import org.apache.zeppelin.notebook.Paragraph; +import org.apache.zeppelin.scheduler.Job; +import org.apache.zeppelin.user.AuthenticationInfo; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Headless implementation of {@link NoteEventListener}. {@link #onParagraphStatusChange} is + * printed to stdout so a blocking CLI run shows per-paragraph progress; the rest are debug-only + * since there is no UI/index to notify in a headless run. + */ +public class HeadlessNoteEventListener implements NoteEventListener { + + private static final Logger LOGGER = LoggerFactory.getLogger(HeadlessNoteEventListener.class); + + @Override + public void onNoteRemove(Note note, AuthenticationInfo subject) { + LOGGER.debug("Note removed: {}", note.getId()); + } + + @Override + public void onNoteCreate(Note note, AuthenticationInfo subject) { + LOGGER.debug("Note created: {}", note.getId()); + } + + @Override + public void onNoteUpdate(Note note, AuthenticationInfo subject) { + LOGGER.debug("Note updated: {}", note.getId()); + } + + @Override + public void onParagraphRemove(Paragraph p) { + LOGGER.debug("Paragraph removed: {}", p.getId()); + } + + @Override + public void onParagraphCreate(Paragraph p) { + LOGGER.debug("Paragraph created: {}", p.getId()); + } + + @Override + public void onParagraphUpdate(Paragraph p) { + LOGGER.debug("Paragraph updated: {}", p.getId()); + } + + @Override + public void onParagraphStatusChange(Paragraph p, Job.Status status) { + System.out.println("[" + p.getId() + "] " + status); + } +} diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/HeadlessProcessListener.java b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/HeadlessProcessListener.java new file mode 100644 index 00000000000..30e2c9e77e3 --- /dev/null +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/HeadlessProcessListener.java @@ -0,0 +1,183 @@ +/* + * 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.notebook.cli; + +import org.apache.thrift.TException; +import org.apache.zeppelin.interpreter.InterpreterResult; +import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcessListener; +import org.apache.zeppelin.interpreter.thrift.ParagraphInfo; +import org.apache.zeppelin.notebook.Notebook; +import org.apache.zeppelin.notebook.Paragraph; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +/** + * Headless implementation of {@link RemoteInterpreterProcessListener}. Only + * {@link #onOutputAppend} and {@link #onOutputUpdated} are forwarded to stdout so a blocking + * CLI run still shows progress; {@link #onOutputClear} and {@link #checkpointOutput} are + * debug-logged only. {@link #getParagraphList} and {@link #runParagraphs} delegate to the + * local {@link Notebook} because some interpreters synchronously depend on them to chain + * paragraph execution (see {@code NotebookServer#runParagraphs}/{@code #getParagraphList} for + * the reference implementation this mirrors, minus the UI-only READER permission check and the + * paragraphIds/paragraphIndices mutual-exclusion guard, both dropped for this single-user + * headless run). + */ +public class HeadlessProcessListener implements RemoteInterpreterProcessListener { + + private static final Logger LOGGER = LoggerFactory.getLogger(HeadlessProcessListener.class); + + private final ExecutorService runParagraphsExecutor = Executors.newCachedThreadPool(r -> { + Thread t = new Thread(r, "HeadlessProcessListener-runParagraphs"); + t.setDaemon(true); + return t; + }); + + private volatile Notebook notebook; + + public void setNotebook(Notebook notebook) { + this.notebook = notebook; + } + + @Override + public void onOutputAppend(String noteId, String paragraphId, int index, String output) { + System.out.print(output); + System.out.flush(); + } + + @Override + public void onOutputUpdated(String noteId, String paragraphId, int index, + InterpreterResult.Type type, String output) { + System.out.println(output); + } + + @Override + public void onOutputClear(String noteId, String paragraphId) { + LOGGER.debug("Output cleared for note {} paragraph {}", noteId, paragraphId); + } + + @Override + public void runParagraphs(String noteId, List paragraphIndices, + List paragraphIds, String curParagraphId) throws IOException { + Notebook nb = requireNotebook(); + nb.processNote(noteId, note -> { + if (note == null) { + throw new IOException("Not existed noteId: " + noteId); + } + List toBeRunParagraphIds = new ArrayList<>(); + if (paragraphIds != null && !paragraphIds.isEmpty()) { + for (String paragraphId : paragraphIds) { + if (note.getParagraph(paragraphId) == null) { + throw new IOException("Not existed paragraphId: " + paragraphId); + } + if (!paragraphId.equals(curParagraphId)) { + toBeRunParagraphIds.add(paragraphId); + } + } + } else if (paragraphIndices != null && !paragraphIndices.isEmpty()) { + for (int paragraphIndex : paragraphIndices) { + Paragraph p = note.getParagraph(paragraphIndex); + if (p == null) { + throw new IOException("Not existed paragraphIndex: " + paragraphIndex); + } + if (!p.getId().equals(curParagraphId)) { + toBeRunParagraphIds.add(p.getId()); + } + } + } else { + for (Paragraph p : note.getParagraphs()) { + if (!p.getId().equals(curParagraphId)) { + toBeRunParagraphIds.add(p.getId()); + } + } + } + runParagraphsExecutor.submit(() -> { + for (String paragraphId : toBeRunParagraphIds) { + try { + note.run(paragraphId, true); + } catch (Exception e) { + LOGGER.warn("Fail to run paragraph {} of note {}", paragraphId, noteId, e); + } + } + }); + return null; + }); + } + + @Override + public void onParaInfosReceived(String noteId, String paragraphId, + String interpreterSettingId, Map metaInfos) { + LOGGER.debug("Paragraph info received for note {} paragraph {}: {}", noteId, paragraphId, + metaInfos); + } + + @Override + public List getParagraphList(String user, String noteId) + throws TException, IOException { + Notebook nb = requireNotebook(); + return nb.processNote(noteId, note -> { + if (note == null) { + throw new IOException("Not found this note: " + noteId); + } + List paragraphInfos = new ArrayList<>(); + for (Paragraph paragraph : note.getParagraphs()) { + ParagraphInfo paraInfo = new ParagraphInfo(); + paraInfo.setNoteId(noteId); + paraInfo.setParagraphId(paragraph.getId()); + paraInfo.setParagraphTitle(paragraph.getTitle()); + paraInfo.setParagraphText(paragraph.getText()); + paragraphInfos.add(paraInfo); + } + return paragraphInfos; + }); + } + + @Override + public void checkpointOutput(String noteId, String paragraphId) { + LOGGER.debug("Checkpoint output for note {} paragraph {}", noteId, paragraphId); + } + + private Notebook requireNotebook() { + Notebook nb = notebook; + if (nb == null) { + throw new IllegalStateException( + "Notebook is not set yet. HeadlessProcessListener.setNotebook must be called " + + "before any interpreter callback can be served."); + } + return nb; + } + + void closeExecutor() { + runParagraphsExecutor.shutdown(); + try { + if (!runParagraphsExecutor.awaitTermination(5, TimeUnit.SECONDS)) { + LOGGER.warn("In-flight runParagraphs task(s) did not finish within 5s, forcing shutdown"); + runParagraphsExecutor.shutdownNow(); + } + } catch (InterruptedException e) { + runParagraphsExecutor.shutdownNow(); + Thread.currentThread().interrupt(); + } + } +} diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/NotebookRunner.java b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/NotebookRunner.java new file mode 100644 index 00000000000..26b361ca961 --- /dev/null +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/NotebookRunner.java @@ -0,0 +1,168 @@ +/* + * 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.notebook.cli; + +import org.apache.zeppelin.conf.ZeppelinConfiguration; +import org.apache.zeppelin.notebook.Note; +import org.apache.zeppelin.notebook.Notebook; +import org.apache.zeppelin.notebook.Paragraph; +import org.apache.zeppelin.scheduler.ExecutorFactory; +import org.apache.zeppelin.scheduler.Job; +import org.apache.zeppelin.user.AuthenticationInfo; +import org.apache.zeppelin.util.IdHashes; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +/** + * CLI entry point for running a Zeppelin note without starting the Zeppelin server (no + * Jetty/REST/WebSocket). Executes every paragraph of the given note via + * {@link Note#runAll}, substituting {@code ${param}} placeholders with values supplied via + * {@code -p}, then saves the executed note back (overwriting the input note, or to a new note + * when {@code -o} is given). Any paragraph left in a non-{@code FINISHED} state (error, abort, + * or skipped because an earlier paragraph failed) fails the run with an exception, so a CI/batch + * caller relying on the process exit code sees the failure. + */ +public final class NotebookRunner { + + private static final Logger LOGGER = LoggerFactory.getLogger(NotebookRunner.class); + + private NotebookRunner() { + } + + public static void main(String[] args) throws Exception { + RunNoteCliOptions options = RunNoteCliOptions.parse(args); + if (options == null) { + // -h/--help was given, usage already printed. + return; + } + + ZeppelinConfiguration zConf = ZeppelinConfiguration.load(); + int exitCode = 0; + // try-with-resources: if run() throws and close() also throws, run()'s exception is the one + // propagated (with close()'s exception attached via Throwable#addSuppressed), instead of + // close() silently masking the real failure. + try (NotebookRunnerContext context = NotebookRunnerContext.bootstrap(zConf)) { + run(context, options); + } catch (Exception e) { + LOGGER.error("Failed to run note", e); + System.err.println("Run failed: " + e.getMessage()); + exitCode = 1; + } + // ExecutorFactory#shutdownAll() belongs here, in main(), rather than in + // NotebookRunnerContext#close(): ExecutorFactory.singleton() is a JVM-wide singleton, not + // owned by any one context, so tearing it down is only correct once this process is truly + // done with it -- exactly the point main() reaches right here. (It must not live in + // close(): that method also runs for every NotebookRunnerContext a test suite creates and + // closes in-process; killing the shared pool there breaks every subsequent test sharing the + // JVM, since e.g. SchedulerFactory lazily creates its backing executor once via + // ExecutorFactory and keeps reusing that same reference for the rest of the JVM's life.) + ExecutorFactory.singleton().shutdownAll(); + // Safety net: everything above already tries to shut down every thread pool it knows about + // (interpreter processes, the event server, RemoteScheduler's own executors, ExecutorFactory's + // named pools) so the JVM can exit on its own. System.exit() here is the backstop for + // whatever it doesn't know about -- some interpreter or future scheduler variant spinning up + // a non-daemon thread this CLI layer has no visibility into -- and it also carries the + // failure signal (HIGH-1) out as a real process exit code for CI/batch callers. + System.exit(exitCode); + } + + static void run(NotebookRunnerContext context, RunNoteCliOptions options) throws IOException { + Notebook notebook = context.getNotebook(); + String noteId = notebook.getNoteIdByPath(options.getNotePath()); + if (noteId == null) { + throw new IOException("Note not found: " + options.getNotePath()); + } + + String outputPath = options.getOutputPath(); + if (outputPath != null) { + if (!outputPath.startsWith("/")) { + throw new IllegalArgumentException( + "-o must be an absolute note path starting with '/': " + outputPath); + } + if (notebook.containsNote(outputPath)) { + throw new IOException("Output note already exists at path: " + outputPath); + } + } + + notebook.processNote(noteId, note -> { + try { + note.runAll(AuthenticationInfo.ANONYMOUS, true, false, options.getParams()); + } catch (Exception e) { + throw new IOException("Failed to run note: " + options.getNotePath(), e); + } + // Save whatever was produced -- including partial results from a run that failed partway + // through -- before deciding whether the run itself should be reported as a failure. + saveResult(notebook, note, options); + // Paragraph output (stdout, via HeadlessProcessListener) may not end with a newline (e.g. + // Python's print(..., end=' ')), so force a line break before the completion summary -- + // otherwise it visually runs into the last paragraph's output on the terminal. + System.out.flush(); + System.err.println(); + System.err.println("Saved executed note to " + note.getPath()); + failIfAnyParagraphDidNotFinish(note); + return null; + }); + } + + private static void saveResult(Notebook notebook, Note note, RunNoteCliOptions options) + throws IOException { + if (options.getOutputPath() != null) { + note.setId(IdHashes.generateId()); + note.setPath(options.getOutputPath()); + } + notebook.saveNote(note, AuthenticationInfo.ANONYMOUS); + } + + /** + * @throws IOException listing every enabled paragraph left in a non-{@code FINISHED} state + * (error, abort, or skipped after an earlier paragraph failed), so a CI/batch caller sees + * a non-zero exit instead of a silently incomplete run. + */ + private static void failIfAnyParagraphDidNotFinish(Note note) throws IOException { + List failedParagraphIds = new ArrayList<>(); + int enabledCount = 0; + for (Paragraph p : note.getParagraphs()) { + if (!p.isEnabled()) { + continue; + } + enabledCount++; + if (p.getStatus() != Job.Status.FINISHED) { + failedParagraphIds.add(p.getId()); + } + } + + if (enabledCount == 0) { + LOGGER.warn("Note {} has no enabled paragraph to run", note.getPath()); + return; + } + + int succeededCount = enabledCount - failedParagraphIds.size(); + LOGGER.info("Note {} finished: {}/{} paragraphs succeeded", note.getPath(), succeededCount, + enabledCount); + System.err.println("Note " + note.getPath() + " finished: " + succeededCount + "/" + + enabledCount + " paragraphs succeeded"); + + if (!failedParagraphIds.isEmpty()) { + throw new IOException("Note " + note.getPath() + " has " + failedParagraphIds.size() + + " failed/incomplete paragraph(s): " + failedParagraphIds); + } + } +} diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/NotebookRunnerContext.java b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/NotebookRunnerContext.java new file mode 100644 index 00000000000..2513604529e --- /dev/null +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/NotebookRunnerContext.java @@ -0,0 +1,169 @@ +/* + * 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.notebook.cli; + +import org.apache.zeppelin.conf.ZeppelinConfiguration; +import org.apache.zeppelin.interpreter.Interpreter; +import org.apache.zeppelin.interpreter.InterpreterFactory; +import org.apache.zeppelin.interpreter.InterpreterSetting; +import org.apache.zeppelin.interpreter.InterpreterSettingManager; +import org.apache.zeppelin.interpreter.ManagedInterpreterGroup; +import org.apache.zeppelin.notebook.AuthorizationService; +import org.apache.zeppelin.notebook.GsonNoteParser; +import org.apache.zeppelin.notebook.NoteManager; +import org.apache.zeppelin.notebook.NoteParser; +import org.apache.zeppelin.notebook.Notebook; +import org.apache.zeppelin.notebook.repo.NotebookRepo; +import org.apache.zeppelin.notebook.repo.VFSNotebookRepo; +import org.apache.zeppelin.plugin.PluginManager; +import org.apache.zeppelin.storage.ConfigStorage; +import org.apache.zeppelin.user.Credentials; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Closeable; +import java.io.IOException; +import java.util.List; +import java.util.concurrent.TimeUnit; + +/** + * Assembles the minimal set of objects needed to run a note headlessly, without Jetty/HK2: a + * production {@link VFSNotebookRepo}-backed {@link Notebook} wired to real (non-mock) listener + * implementations. Follows a manual dependency-injection recipe similar to + * {@code StopInterpreter}/{@code AbstractInterpreterTest}, except it does not separately call + * {@code interpreterSettingManager.setNotebook(notebook)} — {@link Notebook}'s own constructor + * already does that, and the field it populates is only consulted by UI editor-setting/restart + * paths that a headless {@code runAll} never exercises. + */ +public final class NotebookRunnerContext implements Closeable { + + private static final Logger LOGGER = LoggerFactory.getLogger(NotebookRunnerContext.class); + + private final InterpreterSettingManager interpreterSettingManager; + private final InterpreterFactory interpreterFactory; + private final Notebook notebook; + private final HeadlessProcessListener processListener; + + private NotebookRunnerContext(InterpreterSettingManager interpreterSettingManager, + InterpreterFactory interpreterFactory, Notebook notebook, + HeadlessProcessListener processListener) { + this.interpreterSettingManager = interpreterSettingManager; + this.interpreterFactory = interpreterFactory; + this.notebook = notebook; + this.processListener = processListener; + } + + public static NotebookRunnerContext bootstrap(ZeppelinConfiguration zConf) throws IOException { + ConfigStorage storage = ConfigStorage.createConfigStorage(zConf); + PluginManager pluginManager = new PluginManager(zConf); + NoteParser noteParser = new GsonNoteParser(zConf); + + NotebookRepo notebookRepo = new VFSNotebookRepo(); + notebookRepo.init(zConf, noteParser); + NoteManager noteManager = new NoteManager(notebookRepo, zConf); + + HeadlessProcessListener processListener = new HeadlessProcessListener(); + HeadlessAngularObjectListener angularObjectListener = new HeadlessAngularObjectListener(); + HeadlessApplicationEventListener applicationEventListener = + new HeadlessApplicationEventListener(); + + InterpreterSettingManager interpreterSettingManager = new InterpreterSettingManager(zConf, + angularObjectListener, processListener, applicationEventListener, storage, pluginManager); + InterpreterFactory interpreterFactory = new InterpreterFactory(interpreterSettingManager); + + AuthorizationService authorizationService = + new AuthorizationService(noteManager, zConf, storage); + Credentials credentials = new Credentials(zConf, storage); + + Notebook notebook = new Notebook(zConf, authorizationService, notebookRepo, noteManager, + interpreterFactory, interpreterSettingManager, credentials); + notebook.addNotebookEventListener(new HeadlessNoteEventListener()); + processListener.setNotebook(notebook); + + return new NotebookRunnerContext(interpreterSettingManager, interpreterFactory, notebook, + processListener); + } + + public Notebook getNotebook() { + return notebook; + } + + public InterpreterSettingManager getInterpreterSettingManager() { + return interpreterSettingManager; + } + + public InterpreterFactory getInterpreterFactory() { + return interpreterFactory; + } + + HeadlessProcessListener getProcessListener() { + return processListener; + } + + @Override + public void close() throws IOException { + // Order matters: + // 1. Drain our own runParagraphs executor first. + // 2. Stop every RemoteScheduler's *own* job-submission thread pool + // (Executors.newFixedThreadPool/newSingleThreadExecutor named "FIFO-...", created + // directly in RemoteScheduler#createExecutor -- NOT registered in ExecutorFactory) while + // the interpreter groups are still live. InterpreterGroup#close() (invoked below by + // InterpreterSettingManager#close()) only calls Scheduler#stop() -- the no-arg + // overload -- which for RemoteScheduler just interrupts its run() loop and never touches + // that executor field; only the 2-arg Scheduler#stop(timeout, unit) does + // (RemoteScheduler#stop(int, TimeUnit) -> ExecutorUtil.softShutdown). This is a + // pre-existing gap in the shared close() path that a long-lived server never notices + // (it never shuts down), but leaves this "FIFO-RemoteInterpreter-*" thread alive forever + // in a one-shot headless CLI process, hanging the JVM on exit. + // 3. InterpreterSettingManager#close() tears down the interpreter processes themselves + // (they unregister against the event server as they go). + // 4. Stop the event server's own (non-daemon) Thrift server thread -- + // InterpreterSettingManager#close() never does this itself (fine for a long-lived + // server, not for a one-shot CLI process). + // + // NOTE deliberately NOT done here: ExecutorFactory#shutdownAll(). That pool (notably + // SchedulerFactory's backing executor, "SchedulerFactory-*") is a *JVM-wide* singleton + // (ExecutorFactory.singleton()), not owned by this context. Calling shutdownAll() here would + // tear it down for the entire process, including every other NotebookRunnerContext a test + // suite (or any other caller sharing this JVM) creates afterwards -- confirmed by running + // the full test suite: a later test's paragraph submission was rejected with + // RejectedExecutionException because an earlier test's close() had already killed the + // shared pool. NotebookRunner#main calls ExecutorFactory#shutdownAll() itself, once, right + // before the process actually exits -- see its Javadoc for why that's the correct place. + processListener.closeExecutor(); + stopRemoteInterpreterSchedulers(); + interpreterSettingManager.close(); + interpreterSettingManager.getInterpreterEventServer().stop(); + } + + private void stopRemoteInterpreterSchedulers() { + for (InterpreterSetting setting : interpreterSettingManager.get()) { + for (ManagedInterpreterGroup group : setting.getAllInterpreterGroups()) { + for (List session : group.values()) { + for (Interpreter interpreter : session) { + try { + interpreter.getScheduler().stop(5, TimeUnit.SECONDS); + } catch (Exception e) { + LOGGER.warn("Failed to stop scheduler for interpreter {}", + interpreter.getClassName(), e); + } + } + } + } + } + } +} diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/RunNoteCliOptions.java b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/RunNoteCliOptions.java new file mode 100644 index 00000000000..000d9b468e6 --- /dev/null +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/cli/RunNoteCliOptions.java @@ -0,0 +1,123 @@ +/* + * 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.notebook.cli; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * Command line options for {@link NotebookRunner}: {@code -i } (required), + * {@code -o } (optional), {@code -p } (repeatable). Manual switch-based + * parsing, following the same style as {@code InstallInterpreter.main}. + */ +public final class RunNoteCliOptions { + + private static final Logger LOGGER = LoggerFactory.getLogger(RunNoteCliOptions.class); + + private final String notePath; + private final String outputPath; + private final Map params; + + private RunNoteCliOptions(String notePath, String outputPath, Map params) { + this.notePath = notePath; + this.outputPath = outputPath; + this.params = params; + } + + public String getNotePath() { + return notePath; + } + + /** + * @return the output note path, or {@code null} when the input note should be overwritten. + */ + public String getOutputPath() { + return outputPath; + } + + public Map getParams() { + return params; + } + + public static void printUsage() { + System.out.println("Usage: run-note.sh -i [-o ] [-p ]..."); + System.out.println("Options"); + System.out.println(" -i, --input [PATH] Path of the note to run (required)"); + System.out.println(" -o, --output [PATH] Path to save the executed note to. " + + "Defaults to overwriting the input note"); + System.out.println(" -p, --param [KEY] [VALUE] Note parameter, can be repeated"); + System.out.println(" -h, --help Print this help"); + } + + /** + * @return the parsed options, or {@code null} when {@code -h}/{@code --help} was given (usage + * already printed and the caller should exit without running anything). + * @throws IllegalArgumentException when required options are missing or an option is unknown. + */ + public static RunNoteCliOptions parse(String[] args) { + String notePath = null; + String outputPath = null; + Map params = new LinkedHashMap<>(); + + for (int i = 0; i < args.length; i++) { + String arg = args[i]; + switch (arg) { + case "-i": + case "--input": + if (i + 1 >= args.length) { + throw new IllegalArgumentException("Missing value for " + arg); + } + notePath = args[++i]; + break; + case "-o": + case "--output": + if (i + 1 >= args.length) { + throw new IllegalArgumentException("Missing value for " + arg); + } + outputPath = args[++i]; + break; + case "-p": + case "--param": + if (i + 2 >= args.length) { + throw new IllegalArgumentException("Missing key/value for " + arg); + } + String key = args[++i]; + String value = args[++i]; + if (params.containsKey(key)) { + LOGGER.warn("Duplicate -p key '{}': overwriting previous value", key); + } + params.put(key, value); + break; + case "-h": + case "--help": + printUsage(); + return null; + default: + throw new IllegalArgumentException("Unknown option: " + arg); + } + } + + if (notePath == null) { + throw new IllegalArgumentException("-i is required"); + } + + return new RunNoteCliOptions(notePath, outputPath, params); + } +} diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/CliTestFixtures.java b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/CliTestFixtures.java new file mode 100644 index 00000000000..7e0738c551b --- /dev/null +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/CliTestFixtures.java @@ -0,0 +1,90 @@ +/* + * 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.notebook.cli; + +import org.apache.commons.io.FileUtils; +import org.apache.zeppelin.conf.ZeppelinConfiguration; + +import java.io.File; +import java.io.IOException; + +/** + * Shared test fixture setup for the headless {@code notebook.cli} package tests. Mirrors + * {@code AbstractInterpreterTest}'s directory/config bootstrap so each Phase test gets an + * isolated {@code interpreter}/{@code conf}/{@code notebook} directory triad copied from + * {@code src/test/resources}. + */ +final class CliTestFixtures { + + private CliTestFixtures() { + } + + static final class TestDirs { + final File zeppelinHome; + final File interpreterDir; + final File confDir; + final File notebookDir; + final ZeppelinConfiguration zConf; + + private TestDirs(File zeppelinHome, File interpreterDir, File confDir, File notebookDir, + ZeppelinConfiguration zConf) { + this.zeppelinHome = zeppelinHome; + this.interpreterDir = interpreterDir; + this.confDir = confDir; + this.notebookDir = notebookDir; + this.zConf = zConf; + } + } + + static TestDirs setUp(Class testClass) throws IOException { + File zeppelinHome = new File(".."); + File interpreterDir = new File(zeppelinHome, "interpreter_" + testClass.getSimpleName()); + File confDir = new File(zeppelinHome, "conf_" + testClass.getSimpleName()); + File notebookDir = new File(zeppelinHome, "notebook_" + testClass.getSimpleName()); + FileUtils.deleteDirectory(notebookDir); + + interpreterDir.mkdirs(); + confDir.mkdirs(); + notebookDir.mkdirs(); + + FileUtils.copyDirectory(new File("src/test/resources/interpreter"), interpreterDir); + FileUtils.copyDirectory(new File("src/test/resources/conf"), confDir); + + ZeppelinConfiguration zConf = ZeppelinConfiguration.load(); + zConf.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_HOME.getVarName(), + zeppelinHome.getAbsolutePath()); + zConf.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_CONF_DIR.getVarName(), + confDir.getAbsolutePath()); + zConf.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_DIR.getVarName(), + interpreterDir.getAbsolutePath()); + zConf.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTEBOOK_DIR.getVarName(), + notebookDir.getAbsolutePath()); + zConf.setProperty( + ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_GROUP_DEFAULT.getVarName(), "test"); + + return new TestDirs(zeppelinHome, interpreterDir, confDir, notebookDir, zConf); + } + + static void tearDown(TestDirs dirs) throws IOException { + if (dirs == null) { + return; + } + FileUtils.deleteDirectory(dirs.interpreterDir); + FileUtils.deleteDirectory(dirs.confDir); + FileUtils.deleteDirectory(dirs.notebookDir); + } +} diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerContextTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerContextTest.java new file mode 100644 index 00000000000..ece82c06ffa --- /dev/null +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerContextTest.java @@ -0,0 +1,154 @@ +/* + * 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.notebook.cli; + +import org.apache.zeppelin.interpreter.RemoteInterpreterEventServer; +import org.apache.zeppelin.interpreter.thrift.ParagraphInfo; +import org.apache.zeppelin.notebook.Notebook; +import org.apache.zeppelin.notebook.Paragraph; +import org.apache.zeppelin.user.AuthenticationInfo; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import java.io.IOException; +import java.net.Socket; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class NotebookRunnerContextTest { + + private CliTestFixtures.TestDirs dirs; + private NotebookRunnerContext context; + + @BeforeEach + void setUp() throws Exception { + dirs = CliTestFixtures.setUp(NotebookRunnerContextTest.class); + } + + @AfterEach + void tearDown() throws Exception { + if (context != null) { + context.close(); + } + CliTestFixtures.tearDown(dirs); + } + + @Test + @Timeout(30) + void bootstrapProducesWiredComponentsAndClosesCleanly() throws Exception { + context = NotebookRunnerContext.bootstrap(dirs.zConf); + + assertNotNull(context.getNotebook()); + assertNotNull(context.getInterpreterSettingManager()); + assertNotNull(context.getInterpreterFactory()); + + NotebookRunnerContext toClose = context; + context = null; + assertDoesNotThrow(toClose::close); + } + + @Test + @Timeout(30) + void getParagraphListDelegatesToNotebookAndReturnsMatchingInfo() throws Exception { + context = NotebookRunnerContext.bootstrap(dirs.zConf); + Notebook notebook = context.getNotebook(); + String noteId = notebook.createNote("/paragraph-list-note", AuthenticationInfo.ANONYMOUS); + + notebook.processNote(noteId, note -> { + Paragraph p1 = note.addNewParagraph(AuthenticationInfo.ANONYMOUS); + p1.setTitle("first"); + p1.setText("%test.echo hello"); + notebook.saveNote(note, AuthenticationInfo.ANONYMOUS); + return null; + }); + + List paragraphInfos = + context.getProcessListener().getParagraphList("anonymous", noteId); + + assertEquals(1, paragraphInfos.size()); + assertEquals("first", paragraphInfos.get(0).getParagraphTitle()); + assertEquals("%test.echo hello", paragraphInfos.get(0).getParagraphText()); + } + + @Test + @Timeout(30) + void runParagraphsDelegatesToNotebookAndActuallyExecutesThem() throws Exception { + context = NotebookRunnerContext.bootstrap(dirs.zConf); + Notebook notebook = context.getNotebook(); + String noteId = notebook.createNote("/run-paragraphs-note", AuthenticationInfo.ANONYMOUS); + + String paragraphId = notebook.processNote(noteId, note -> { + Paragraph p1 = note.addNewParagraph(AuthenticationInfo.ANONYMOUS); + p1.setText("%mock1 hello-from-run-paragraphs"); + notebook.saveNote(note, AuthenticationInfo.ANONYMOUS); + return p1.getId(); + }); + + context.getProcessListener() + .runParagraphs(noteId, Collections.emptyList(), Collections.emptyList(), ""); + + await().atMost(15, TimeUnit.SECONDS).until(() -> notebook.processNote(noteId, + note -> note.getParagraph(paragraphId).getReturn() != null)); + + notebook.processNote(noteId, note -> { + assertEquals("repl1: hello-from-run-paragraphs", + note.getParagraph(paragraphId).getReturn().message().get(0).getData()); + return null; + }); + } + + @Test + @Timeout(15) + void closeStopsTheEventServerSoItNoLongerAcceptsConnections() throws Exception { + context = NotebookRunnerContext.bootstrap(dirs.zConf); + RemoteInterpreterEventServer eventServer = + context.getInterpreterSettingManager().getInterpreterEventServer(); + String host = eventServer.getHost(); + int port = eventServer.getPort(); + + // Sanity check: the event server is actually listening before close() -- otherwise the + // "connection refused after close()" assertion below would pass for the wrong reason. + try (Socket before = new Socket(host, port)) { + assertTrue(before.isConnected()); + } + + NotebookRunnerContext toClose = context; + context = null; + toClose.close(); + + // close() must have stopped RemoteInterpreterEventServer's (non-daemon) Thrift server + // thread -- otherwise that thread keeps the JVM alive forever after main() returns. Proven + // black-box here (no accessor for the server's internal isServing() state) by observing + // that the port it was listening on now refuses connections. + await().atMost(5, TimeUnit.SECONDS).until(() -> { + try (Socket after = new Socket(host, port)) { + return false; + } catch (IOException e) { + return true; + } + }); + } +} diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerIntegrationTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerIntegrationTest.java new file mode 100644 index 00000000000..5d24e955898 --- /dev/null +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerIntegrationTest.java @@ -0,0 +1,225 @@ +/* + * 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.notebook.cli; + +import org.apache.zeppelin.conf.ZeppelinConfiguration; +import org.apache.zeppelin.notebook.Notebook; +import org.apache.zeppelin.notebook.Paragraph; +import org.apache.zeppelin.user.AuthenticationInfo; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import java.io.BufferedReader; +import java.io.File; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Phase 5, the completion-criteria proof: drives the CLI end to end as a real, separate JVM + * process via {@code bin/run-note.sh} (no server, no mocked note/paragraph) and asserts on the + * one thing an in-process call to {@link NotebookRunner#main} can never observe: whether the + * process actually exits. + * + *

This has to run out-of-process for two independent reasons: (1) proving there's no JVM + * hang requires watching an external process die on its own -- a same-JVM call can't see that, + * the surefire JVM is alive for its own reasons regardless of what {@code close()} did; and (2) + * {@link NotebookRunner#main} now calls {@code System.exit()} as a last-resort cleanup net, which + * would kill the test runner itself if called in-process. + */ +class NotebookRunnerIntegrationTest { + + private static final String REMOTE_INTERPRETER_SERVER_CLASS = + "org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer"; + + private CliTestFixtures.TestDirs dirs; + + @BeforeEach + void setUp() throws Exception { + dirs = CliTestFixtures.setUp(NotebookRunnerIntegrationTest.class); + System.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_HOME.getVarName(), + dirs.zeppelinHome.getAbsolutePath()); + System.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_CONF_DIR.getVarName(), + dirs.confDir.getAbsolutePath()); + System.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_DIR.getVarName(), + dirs.interpreterDir.getAbsolutePath()); + System.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTEBOOK_DIR.getVarName(), + dirs.notebookDir.getAbsolutePath()); + System.setProperty( + ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_GROUP_DEFAULT.getVarName(), "test"); + + // Fixture notes, written directly through a throwaway context so the subprocess only has to + // run them. + try (NotebookRunnerContext setupContext = NotebookRunnerContext.bootstrap(dirs.zConf)) { + Notebook notebook = setupContext.getNotebook(); + + String successNoteId = notebook.createNote("/integration-note", AuthenticationInfo.ANONYMOUS); + notebook.processNote(successNoteId, note -> { + Paragraph p = note.addNewParagraph(AuthenticationInfo.ANONYMOUS); + p.setText("%mock1 ${msg=default}"); + notebook.saveNote(note, AuthenticationInfo.ANONYMOUS); + return null; + }); + + String failureNoteId = + notebook.createNote("/integration-failure-note", AuthenticationInfo.ANONYMOUS); + notebook.processNote(failureNoteId, note -> { + Paragraph p = note.addNewParagraph(AuthenticationInfo.ANONYMOUS); + p.setText("%nonexistent boom"); + notebook.saveNote(note, AuthenticationInfo.ANONYMOUS); + return null; + }); + } + } + + @AfterEach + void tearDown() throws Exception { + System.clearProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_HOME.getVarName()); + System.clearProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_CONF_DIR.getVarName()); + System.clearProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_DIR.getVarName()); + System.clearProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTEBOOK_DIR.getVarName()); + System.clearProperty( + ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_GROUP_DEFAULT.getVarName()); + CliTestFixtures.tearDown(dirs); + } + + @Test + @Timeout(90) + void mainExitsZeroSavesResultAndLeavesNoOrphanProcessWhenNoteSucceeds() throws Exception { + SubprocessResult result = + runNoteScript("-i", "/integration-note", "-p", "msg", "subprocess-hello"); + + assertTrue(result.exited, + "run-note.sh subprocess did not exit within 60s -- JVM hang. Output so far:\n" + + result.output); + assertEquals(0, result.exitCode, "Unexpected exit code. Output:\n" + result.output); + + // Verify the saved note independently, through a fresh context. + try (NotebookRunnerContext verifyContext = NotebookRunnerContext.bootstrap(dirs.zConf)) { + Notebook notebook = verifyContext.getNotebook(); + String noteId = notebook.getNoteIdByPath("/integration-note"); + notebook.processNote(noteId, true, note -> { + assertEquals("repl1: subprocess-hello", + note.getParagraphs().get(0).getReturn().message().get(0).getData()); + return null; + }); + } + + // Confirm at the OS level (not just in-JVM bookkeeping) that no interpreter subprocess + // survived the parent CLI process exiting. + assertEquals(0, countRunningRemoteInterpreterServerProcesses()); + } + + @Test + @Timeout(90) + void mainExitsOneWhenParagraphFails() throws Exception { + SubprocessResult result = runNoteScript("-i", "/integration-failure-note"); + + assertTrue(result.exited, + "run-note.sh subprocess did not exit within 60s -- JVM hang. Output so far:\n" + + result.output); + assertEquals(1, result.exitCode, "Unexpected exit code. Output:\n" + result.output); + } + + private static final class SubprocessResult { + final boolean exited; + final int exitCode; + final String output; + + SubprocessResult(boolean exited, int exitCode, String output) { + this.exited = exited; + this.exitCode = exitCode; + this.output = output; + } + } + + /** + * Runs {@code bin/run-note.sh} as a genuinely separate JVM process against this test's fixture + * dirs. Uses the script rather than a raw {@code java -cp ...} child process because surefire's + * default fork mode uses a manifest-only jar for its own classpath -- {@code + * System.getProperty("java.class.path")} inside this test JVM would not be directly reusable + * for a child {@code java -cp} invocation, while the script assembles its own classpath from + * {@code ZEPPELIN_HOME}. + */ + private SubprocessResult runNoteScript(String... args) throws Exception { + File runNoteScript = new File(dirs.zeppelinHome, "bin/run-note.sh"); + assertTrue(runNoteScript.isFile(), "bin/run-note.sh not found at " + runNoteScript); + + java.util.List command = new java.util.ArrayList<>(); + command.add(runNoteScript.getAbsolutePath()); + command.addAll(Arrays.asList(args)); + + ProcessBuilder pb = new ProcessBuilder(command); + pb.redirectErrorStream(true); + pb.environment().put("ZEPPELIN_HOME", dirs.zeppelinHome.getAbsolutePath()); + pb.environment().put("ZEPPELIN_CONF_DIR", dirs.confDir.getAbsolutePath()); + pb.environment().put("ZEPPELIN_LOG_DIR", + new File(dirs.zeppelinHome, "logs_" + getClass().getSimpleName()).getAbsolutePath()); + pb.environment().put("ZEPPELIN_JAVA_OPTS", + "-Dzeppelin.home=" + dirs.zeppelinHome.getAbsolutePath() + + " -Dzeppelin.conf.dir=" + dirs.confDir.getAbsolutePath() + + " -Dzeppelin.interpreter.dir=" + dirs.interpreterDir.getAbsolutePath() + + " -Dzeppelin.notebook.dir=" + dirs.notebookDir.getAbsolutePath() + + " -Dzeppelin.interpreter.group.default=test"); + + Process process = pb.start(); + StringBuilder output = new StringBuilder(); + Thread drain = new Thread(() -> { + try (BufferedReader reader = new BufferedReader( + new InputStreamReader(process.getInputStream(), StandardCharsets.UTF_8))) { + String line; + while ((line = reader.readLine()) != null) { + output.append(line).append(System.lineSeparator()); + } + } catch (Exception e) { + // best-effort draining only + } + }); + drain.setDaemon(true); + drain.start(); + + boolean exited = process.waitFor(60, TimeUnit.SECONDS); + if (!exited) { + process.destroyForcibly(); + } + drain.join(TimeUnit.SECONDS.toMillis(5)); + + return new SubprocessResult(exited, exited ? process.exitValue() : -1, output.toString()); + } + + private static int countRunningRemoteInterpreterServerProcesses() throws Exception { + Process jps = new ProcessBuilder("jps", "-l").start(); + int count = 0; + try (BufferedReader reader = new BufferedReader( + new InputStreamReader(jps.getInputStream(), StandardCharsets.UTF_8))) { + String line; + while ((line = reader.readLine()) != null) { + if (line.contains(REMOTE_INTERPRETER_SERVER_CLASS)) { + count++; + } + } + } + jps.waitFor(); + return count; + } +} diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerOutputSaveTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerOutputSaveTest.java new file mode 100644 index 00000000000..649b510aedc --- /dev/null +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerOutputSaveTest.java @@ -0,0 +1,168 @@ +/* + * 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.notebook.cli; + +import org.apache.zeppelin.notebook.Notebook; +import org.apache.zeppelin.notebook.Paragraph; +import org.apache.zeppelin.user.AuthenticationInfo; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import java.io.IOException; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * Phase 3: verifies the input=output (overwrite) and input!=output (new note, original + * untouched) save branches. Reloads notes with {@code reload=true} to bypass the in-process + * note cache and prove the assertions hold against what {@code VFSNotebookRepo} actually wrote + * to disk, not just in-memory state. + */ +class NotebookRunnerOutputSaveTest { + + private CliTestFixtures.TestDirs dirs; + private NotebookRunnerContext context; + + @BeforeEach + void setUp() throws Exception { + dirs = CliTestFixtures.setUp(NotebookRunnerOutputSaveTest.class); + context = NotebookRunnerContext.bootstrap(dirs.zConf); + } + + @AfterEach + void tearDown() throws Exception { + context.close(); + CliTestFixtures.tearDown(dirs); + } + + private String createNoteWithParagraph(String path, String scriptText) throws Exception { + Notebook notebook = context.getNotebook(); + String noteId = notebook.createNote(path, AuthenticationInfo.ANONYMOUS); + notebook.processNote(noteId, note -> { + Paragraph p = note.addNewParagraph(AuthenticationInfo.ANONYMOUS); + p.setText(scriptText); + notebook.saveNote(note, AuthenticationInfo.ANONYMOUS); + return null; + }); + return noteId; + } + + @Test + @Timeout(30) + void noOutputPathOverwritesTheInputNoteInPlace() throws Exception { + Notebook notebook = context.getNotebook(); + String noteId = createNoteWithParagraph("/save-inplace-note", "%mock1 in-place"); + + NotebookRunner.run(context, RunNoteCliOptions.parse(new String[] {"-i", "/save-inplace-note"})); + + // Reload from disk (bypass cache) to prove the result was actually persisted. + notebook.processNote(noteId, true, note -> { + assertEquals(noteId, note.getId()); + assertEquals("/save-inplace-note", note.getPath()); + assertEquals("repl1: in-place", + note.getParagraphs().get(0).getReturn().message().get(0).getData()); + return null; + }); + assertEquals(noteId, notebook.getNoteIdByPath("/save-inplace-note")); + } + + @Test + @Timeout(30) + void outputPathSavesANewNoteAndLeavesTheInputNoteUntouched() throws Exception { + Notebook notebook = context.getNotebook(); + String inputNoteId = createNoteWithParagraph("/save-output-input", "%mock1 to-output"); + + NotebookRunner.run(context, RunNoteCliOptions.parse( + new String[] {"-i", "/save-output-input", "-o", "/save-output-result"})); + + String outputNoteId = notebook.getNoteIdByPath("/save-output-result"); + assertNotEquals(inputNoteId, outputNoteId); + + notebook.processNote(outputNoteId, true, note -> { + assertEquals("repl1: to-output", + note.getParagraphs().get(0).getReturn().message().get(0).getData()); + return null; + }); + + // Original input note, reloaded fresh from disk, must be untouched (no result attached). + notebook.processNote(inputNoteId, true, note -> { + assertEquals("/save-output-input", note.getPath()); + assertNull(note.getParagraphs().get(0).getReturn()); + return null; + }); + } + + @Test + @Timeout(30) + void runThrowsWhenParagraphFailsButSavesPartialResult() throws Exception { + Notebook notebook = context.getNotebook(); + String noteId = notebook.createNote("/save-failure-note", AuthenticationInfo.ANONYMOUS); + notebook.processNote(noteId, note -> { + Paragraph ok = note.addNewParagraph(AuthenticationInfo.ANONYMOUS); + ok.setText("%mock1 succeeds"); + Paragraph bad = note.addNewParagraph(AuthenticationInfo.ANONYMOUS); + bad.setText("%nonexistent boom"); + notebook.saveNote(note, AuthenticationInfo.ANONYMOUS); + return null; + }); + + RunNoteCliOptions options = RunNoteCliOptions.parse(new String[] {"-i", "/save-failure-note"}); + assertThrows(IOException.class, () -> NotebookRunner.run(context, options)); + + // Even though the run failed overall, the successful paragraph's result must have been + // persisted -- a failing note must not silently lose the work it did complete. + notebook.processNote(noteId, true, note -> { + assertEquals("repl1: succeeds", + note.getParagraphs().get(0).getReturn().message().get(0).getData()); + return null; + }); + } + + @Test + @Timeout(30) + void outputPathCollisionWithExistingNoteThrows() throws Exception { + createNoteWithParagraph("/save-collision-input", "%mock1 input"); + createNoteWithParagraph("/save-collision-existing", "%mock1 existing"); + + RunNoteCliOptions options = RunNoteCliOptions.parse( + new String[] {"-i", "/save-collision-input", "-o", "/save-collision-existing"}); + + assertThrows(IOException.class, () -> NotebookRunner.run(context, options)); + } + + @Test + @Timeout(30) + void outputPathWithoutLeadingSlashThrowsBeforeExecution() throws Exception { + String noteId = createNoteWithParagraph("/save-noslash-input", "%mock1 input"); + + RunNoteCliOptions options = RunNoteCliOptions.parse( + new String[] {"-i", "/save-noslash-input", "-o", "no-leading-slash"}); + + assertThrows(IllegalArgumentException.class, () -> NotebookRunner.run(context, options)); + + // Must have failed before execution: no result attached to the paragraph. + context.getNotebook().processNote(noteId, true, note -> { + assertNull(note.getParagraphs().get(0).getReturn()); + return null; + }); + } +} diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerParamSubstitutionTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerParamSubstitutionTest.java new file mode 100644 index 00000000000..aef2dd8a76e --- /dev/null +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerParamSubstitutionTest.java @@ -0,0 +1,137 @@ +/* + * 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.notebook.cli; + +import org.apache.zeppelin.notebook.Notebook; +import org.apache.zeppelin.notebook.Paragraph; +import org.apache.zeppelin.user.AuthenticationInfo; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.apache.zeppelin.scheduler.Job.Status.ERROR; + +class NotebookRunnerParamSubstitutionTest { + + private CliTestFixtures.TestDirs dirs; + private NotebookRunnerContext context; + + @BeforeEach + void setUp() throws Exception { + dirs = CliTestFixtures.setUp(NotebookRunnerParamSubstitutionTest.class); + context = NotebookRunnerContext.bootstrap(dirs.zConf); + } + + @AfterEach + void tearDown() throws Exception { + context.close(); + CliTestFixtures.tearDown(dirs); + } + + private String createNoteWithParagraph(String path, String scriptText) throws Exception { + Notebook notebook = context.getNotebook(); + String noteId = notebook.createNote(path, AuthenticationInfo.ANONYMOUS); + notebook.processNote(noteId, note -> { + Paragraph p = note.addNewParagraph(AuthenticationInfo.ANONYMOUS); + p.setText(scriptText); + notebook.saveNote(note, AuthenticationInfo.ANONYMOUS); + return null; + }); + return noteId; + } + + @Test + @Timeout(30) + void explicitParamOverridesDefaultValue() throws Exception { + String noteId = createNoteWithParagraph("/param-note-1", "%mock1 Hello ${name=World}"); + + RunNoteCliOptions options = RunNoteCliOptions.parse( + new String[] {"-i", "/param-note-1", "-p", "name", "Zeppelin"}); + NotebookRunner.run(context, options); + + context.getNotebook().processNote(noteId, note -> { + Paragraph p = note.getParagraphs().get(0); + assertNotEquals(ERROR, p.getStatus()); + assertEquals("repl1: Hello Zeppelin", p.getReturn().message().get(0).getData()); + return null; + }); + } + + @Test + @Timeout(30) + void missingParamFallsBackToDefaultValue() throws Exception { + String noteId = createNoteWithParagraph("/param-note-2", "%mock1 Hello ${name=World}"); + + RunNoteCliOptions options = RunNoteCliOptions.parse(new String[] {"-i", "/param-note-2"}); + NotebookRunner.run(context, options); + + context.getNotebook().processNote(noteId, note -> { + Paragraph p = note.getParagraphs().get(0); + assertNotEquals(ERROR, p.getStatus()); + assertEquals("repl1: Hello World", p.getReturn().message().get(0).getData()); + return null; + }); + } + + @Test + @Timeout(30) + void repeatedParamOptionsSubstituteAllKeys() throws Exception { + String noteId = createNoteWithParagraph("/param-note-3", + "%mock1 ${greeting=Hi} ${name=World}"); + + RunNoteCliOptions options = RunNoteCliOptions.parse(new String[] { + "-i", "/param-note-3", + "-p", "greeting", "Hello", + "-p", "name", "Zeppelin"}); + NotebookRunner.run(context, options); + + context.getNotebook().processNote(noteId, note -> { + Paragraph p = note.getParagraphs().get(0); + assertNotEquals(ERROR, p.getStatus()); + assertEquals("repl1: Hello Zeppelin", p.getReturn().message().get(0).getData()); + return null; + }); + } + + @Test + void parseFailsFastWhenNotePathMissing() { + assertThrows(IllegalArgumentException.class, + () -> RunNoteCliOptions.parse(new String[] {"-p", "name", "Zeppelin"})); + } + + @Test + void parseFailsFastWhenInputValueMissing() { + assertThrows(IllegalArgumentException.class, + () -> RunNoteCliOptions.parse(new String[] {"-i"})); + } + + @Test + void parseFailsFastWhenOutputValueMissing() { + assertThrows(IllegalArgumentException.class, + () -> RunNoteCliOptions.parse(new String[] {"-i", "n.zpln", "-o"})); + } + + @Test + void parseFailsFastWhenParamValueMissing() { + assertThrows(IllegalArgumentException.class, + () -> RunNoteCliOptions.parse(new String[] {"-i", "n.zpln", "-p", "key"})); + } +} diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerPrototypeTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerPrototypeTest.java new file mode 100644 index 00000000000..fda563d2474 --- /dev/null +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerPrototypeTest.java @@ -0,0 +1,89 @@ +/* + * 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.notebook.cli; + +import org.apache.zeppelin.display.AngularObjectRegistryListener; +import org.apache.zeppelin.helium.ApplicationEventListener; +import org.apache.zeppelin.interpreter.ExecutionContext; +import org.apache.zeppelin.interpreter.Interpreter; +import org.apache.zeppelin.interpreter.InterpreterContext; +import org.apache.zeppelin.interpreter.InterpreterFactory; +import org.apache.zeppelin.interpreter.InterpreterResult; +import org.apache.zeppelin.interpreter.InterpreterSettingManager; +import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcessListener; +import org.apache.zeppelin.plugin.PluginManager; +import org.apache.zeppelin.storage.ConfigStorage; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; + +/** + * Phase 0 gate: proves that a headless bootstrap built the exact way + * {@code StopInterpreter} assembles {@link InterpreterSettingManager} (no Jetty/HK2 involved) + * can actually launch an interpreter process and receive its Thrift registration callback, + * by observing a synchronous {@link InterpreterResult.Code#SUCCESS} from + * {@link Interpreter#interpret}. + * + *

If this test times out or hangs, the headless approach itself is broken (port binding or + * process callback failure) and Phase 1 onward must not proceed. + */ +class NotebookRunnerPrototypeTest { + + private CliTestFixtures.TestDirs dirs; + private InterpreterSettingManager interpreterSettingManager; + + @BeforeEach + void setUp() throws Exception { + dirs = CliTestFixtures.setUp(NotebookRunnerPrototypeTest.class); + } + + @AfterEach + void tearDown() throws Exception { + if (interpreterSettingManager != null) { + interpreterSettingManager.close(); + } + CliTestFixtures.tearDown(dirs); + } + + @Test + @Timeout(30) + void interpretReturnsSuccessSynchronouslyThroughRealInterpreterProcess() throws Exception { + ConfigStorage storage = ConfigStorage.createConfigStorage(dirs.zConf); + PluginManager pluginManager = new PluginManager(dirs.zConf); + interpreterSettingManager = new InterpreterSettingManager(dirs.zConf, + mock(AngularObjectRegistryListener.class), + mock(RemoteInterpreterProcessListener.class), + mock(ApplicationEventListener.class), + storage, pluginManager); + InterpreterFactory interpreterFactory = new InterpreterFactory(interpreterSettingManager); + + Interpreter interpreter = interpreterFactory.getInterpreter("", + new ExecutionContext("user1", "noteId", "test")); + InterpreterContext context = InterpreterContext.builder() + .setNoteId("noteId") + .setParagraphId("paragraphId") + .build(); + + InterpreterResult result = interpreter.interpret("echo hello", context); + + assertEquals(InterpreterResult.Code.SUCCESS, result.code()); + } +} From 4dff89b711d74f2cb2884a10da71fcbc768d908a Mon Sep 17 00:00:00 2001 From: HwangRock Date: Sun, 6 Sep 2026 00:17:56 +0900 Subject: [PATCH 2/2] [ZEPPELIN-5745] Scope run-note orphan check to the spawned subprocess tree NotebookRunnerIntegrationTest counted RemoteInterpreterServer JVMs machine-wide via `jps -l`, so any concurrent zeppelin-server test that spawned its own interpreter made the orphan assertion see non-zero survivors (expected:0 but was:8 in the core-modules CI job). Poll the descendant pids of the run-note.sh process while it is alive and assert none of them survive the parent exit, scoping the check to this test's own process tree. --- .../cli/NotebookRunnerIntegrationTest.java | 70 +++++++++++++------ 1 file changed, 47 insertions(+), 23 deletions(-) diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerIntegrationTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerIntegrationTest.java index 5d24e955898..50d8bac5c2a 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerIntegrationTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/cli/NotebookRunnerIntegrationTest.java @@ -30,7 +30,12 @@ import java.io.InputStreamReader; import java.nio.charset.StandardCharsets; import java.util.Arrays; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -49,9 +54,6 @@ */ class NotebookRunnerIntegrationTest { - private static final String REMOTE_INTERPRETER_SERVER_CLASS = - "org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer"; - private CliTestFixtures.TestDirs dirs; @BeforeEach @@ -126,8 +128,11 @@ void mainExitsZeroSavesResultAndLeavesNoOrphanProcessWhenNoteSucceeds() throws E } // Confirm at the OS level (not just in-JVM bookkeeping) that no interpreter subprocess - // survived the parent CLI process exiting. - assertEquals(0, countRunningRemoteInterpreterServerProcesses()); + // survived the parent CLI process exiting. orphanPids is every descendant pid observed + // under the run-note.sh process (which includes the interpreter subprocess it spawns) + // that is still alive now that the parent has exited. + assertTrue(result.orphanPids.isEmpty(), + "Orphan descendant process(es) survived parent exit: " + result.orphanPids); } @Test @@ -145,11 +150,18 @@ private static final class SubprocessResult { final boolean exited; final int exitCode; final String output; - - SubprocessResult(boolean exited, int exitCode, String output) { + /** + * Descendant pids of the run-note.sh process (collected while it was still alive, since + * {@link Process#descendants()} stops reporting anything useful once the parent has + * terminated) that are still alive now that the parent has exited. + */ + final List orphanPids; + + SubprocessResult(boolean exited, int exitCode, String output, List orphanPids) { this.exited = exited; this.exitCode = exitCode; this.output = output; + this.orphanPids = orphanPids; } } @@ -198,28 +210,40 @@ private SubprocessResult runNoteScript(String... args) throws Exception { drain.setDaemon(true); drain.start(); + // run-note.sh forks its own java child (NotebookRunner), which in turn forks the + // interpreter subprocess (RemoteInterpreterServer) -- both are descendants of `process`. + // Process#descendants() only reliably walks that tree while the parent is still alive, so + // we have to poll and union pids while run-note.sh is running rather than snapshot once + // after it exits. + Set descendantPids = ConcurrentHashMap.newKeySet(); + AtomicBoolean stopPolling = new AtomicBoolean(false); + Thread descendantPoller = new Thread(() -> { + while (!stopPolling.get()) { + process.descendants().map(ProcessHandle::pid).forEach(descendantPids::add); + try { + Thread.sleep(50); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + } + }); + descendantPoller.setDaemon(true); + descendantPoller.start(); + boolean exited = process.waitFor(60, TimeUnit.SECONDS); if (!exited) { process.destroyForcibly(); } + stopPolling.set(true); + descendantPoller.join(TimeUnit.SECONDS.toMillis(5)); drain.join(TimeUnit.SECONDS.toMillis(5)); - return new SubprocessResult(exited, exited ? process.exitValue() : -1, output.toString()); - } + List orphanPids = descendantPids.stream() + .filter(pid -> ProcessHandle.of(pid).map(ProcessHandle::isAlive).orElse(false)) + .collect(Collectors.toList()); - private static int countRunningRemoteInterpreterServerProcesses() throws Exception { - Process jps = new ProcessBuilder("jps", "-l").start(); - int count = 0; - try (BufferedReader reader = new BufferedReader( - new InputStreamReader(jps.getInputStream(), StandardCharsets.UTF_8))) { - String line; - while ((line = reader.readLine()) != null) { - if (line.contains(REMOTE_INTERPRETER_SERVER_CLASS)) { - count++; - } - } - } - jps.waitFor(); - return count; + return new SubprocessResult( + exited, exited ? process.exitValue() : -1, output.toString(), orphanPids); } }