From 8367acfc0526965a884f58eea8f8ddda70d011d7 Mon Sep 17 00:00:00 2001 From: DuyHai DOAN Date: Thu, 2 Jun 2016 16:15:46 +0200 Subject: [PATCH 1/2] [ZEPPELIN-699] Add new synchronous paragraph run REST API --- .../apache/zeppelin/rest/NotebookRestApi.java | 67 +++++++++++++++++++ 1 file changed, 67 insertions(+) diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java index 6a286a4e19c..d26d30a7e86 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java @@ -609,6 +609,58 @@ public Response runParagraph(@PathParam("notebookId") String notebookId, return new JsonResponse<>(Status.OK).build(); } +/** + * Run synchronously a paragraph REST API + * + * @param noteId - noteId + * @param paragraphId - paragraphId + * @param message - JSON with params if user wants to update dynamic form's value + * null, empty string, empty json if user doesn't want to update + * + * @return JSON with status.OK + * @throws IOException, IllegalArgumentException + */ + @POST + @Path("run/{notebookId}/{paragraphId}") + @ZeppelinApi + public Response runParagraphSynchronously(@PathParam("notebookId") String noteId, + @PathParam("paragraphId") String paragraphId, + String message) throws + IOException, IllegalArgumentException { + LOG.info("run paragraph synchronously {} {} {}", noteId, paragraphId, message); + + Note note = notebook.getNote(noteId); + if (note == null) { + return new JsonResponse<>(Status.NOT_FOUND, "note not found.").build(); + } + + Paragraph paragraph = note.getParagraph(paragraphId); + if (paragraph == null) { + return new JsonResponse<>(Status.NOT_FOUND, "paragraph not found.").build(); + } + + // handle params if presented + handleParagraphParams(message, note, paragraph); + + if (paragraph.getNoteReplLoader() == null) { + paragraph.setNoteReplLoader(note.getNoteReplLoader()); + } + + if (paragraph.getListener() == null) { + paragraph.setListener(note.getJobListenerFactory().getParagraphJobListener(note)); + } + + paragraph.run(); + + final InterpreterResult result = paragraph.getResult(); + + if (result.code() == InterpreterResult.Code.SUCCESS) { + return new JsonResponse<>(Status.OK, result).build(); + } else { + return new JsonResponse<>(Status.INTERNAL_SERVER_ERROR, result).build(); + } + } + /** * Stop(delete) paragraph job REST API * @@ -800,4 +852,19 @@ public Response search(@QueryParam("q") String queryTerm) { return new JsonResponse<>(Status.OK, notebooksFound).build(); } + + private void handleParagraphParams(String message, Note note, Paragraph paragraph) + throws IOException { + // handle params if presented + if (!StringUtils.isEmpty(message)) { + RunParagraphWithParametersRequest request = gson.fromJson(message, + RunParagraphWithParametersRequest.class); + Map paramsForUpdating = request.getParams(); + if (paramsForUpdating != null) { + paragraph.settings.getParams().putAll(paramsForUpdating); + note.persist(); + } + } + } + } From fb0570ce70c47d3648ce4b9cc8265ebe51f6bdc8 Mon Sep 17 00:00:00 2001 From: DuyHai DOAN Date: Thu, 2 Jun 2016 16:24:38 +0200 Subject: [PATCH 2/2] [ZEPPELIN-699] Update Notebook REST API documentation --- docs/rest-api/rest-notebook.md | 55 ++++++++++++++++++- .../apache/zeppelin/rest/NotebookRestApi.java | 30 +++------- .../org/apache/zeppelin/notebook/Note.java | 26 ++++++++- 3 files changed, 86 insertions(+), 25 deletions(-) diff --git a/docs/rest-api/rest-notebook.md b/docs/rest-api/rest-notebook.md index c7e17ea77d0..ed39a4ff8f7 100644 --- a/docs/rest-api/rest-notebook.md +++ b/docs/rest-api/rest-notebook.md @@ -450,12 +450,12 @@ If you work with Apache Zeppelin and find a need for an additional REST API, ple
-### Run a paragraph +### Run a paragraph asynchronously - @@ -487,6 +487,56 @@ If you work with Apache Zeppelin and find a need for an additional REST API, ple
DescriptionThis ```POST``` method runs the paragraph by given notebook and paragraph id. + This ```POST``` method runs the paragraph asynchronously by given notebook and paragraph id. This API always return SUCCESS even if the execution of the paragraph fails later because the API is asynchronous
+
+### Run a paragraph synchronously + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
Description This ```POST``` method runs the paragraph synchronously by given notebook and paragraph id. This API can return SUCCESS or ERROR depending on the outcome of the paragraph execution +
URL```http://[zeppelin-server]:[zeppelin-port]/api/notebook/job/[notebookId]/[paragraphId]```
Success code200
Fail code 500
sample JSON input (optional, only needed when if you want to update dynamic form's value) 
+{
+  "name": "name of new notebook",
+  "params": {
+    "formLabel1": "value1",
+    "formLabel2": "value2"
+  }
+}
sample JSON response 
{"status": "OK"}
sample JSON error 
+{
+   "status": "INTERNAL\_SERVER\_ERROR",
+   "body": {
+       "code": "ERROR",
+       "type": "TEXT",
+       "msg": "bash: -c: line 0: unexpected EOF while looking for matching ``'\nbash: -c: line 1: syntax error: unexpected end of file\nExitValue: 2"
+   }
+}
+
### Stop a paragraph @@ -922,4 +972,3 @@ If you work with Apache Zeppelin and find a need for an additional REST API, ple
- \ No newline at end of file diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java index d26d30a7e86..d148428e498 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java @@ -567,7 +567,7 @@ public Response getNoteParagraphJobStatus(@PathParam("notebookId") String notebo } /** - * Run paragraph job REST API + * Run asynchronously paragraph job REST API * * @param message - JSON with params if user wants to update dynamic form's value * null, empty string, empty json if user doesn't want to update @@ -580,7 +580,7 @@ public Response getNoteParagraphJobStatus(@PathParam("notebookId") String notebo public Response runParagraph(@PathParam("notebookId") String notebookId, @PathParam("paragraphId") String paragraphId, String message) throws IOException, IllegalArgumentException { - LOG.info("run paragraph job {} {} {}", notebookId, paragraphId, message); + LOG.info("run paragraph job asynchronously {} {} {}", notebookId, paragraphId, message); Note note = notebook.getNote(notebookId); if (note == null) { @@ -593,17 +593,7 @@ public Response runParagraph(@PathParam("notebookId") String notebookId, } // handle params if presented - if (!StringUtils.isEmpty(message)) { - RunParagraphWithParametersRequest request = - gson.fromJson(message, RunParagraphWithParametersRequest.class); - Map paramsForUpdating = request.getParams(); - if (paramsForUpdating != null) { - paragraph.settings.getParams().putAll(paramsForUpdating); - AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal()); - note.setLastReplName(paragraph.getId()); - note.persist(subject); - } - } + handleParagraphParams(message, note, paragraph); note.run(paragraph.getId()); return new JsonResponse<>(Status.OK).build(); @@ -642,12 +632,8 @@ public Response runParagraphSynchronously(@PathParam("notebookId") String noteId // handle params if presented handleParagraphParams(message, note, paragraph); - if (paragraph.getNoteReplLoader() == null) { - paragraph.setNoteReplLoader(note.getNoteReplLoader()); - } - if (paragraph.getListener() == null) { - paragraph.setListener(note.getJobListenerFactory().getParagraphJobListener(note)); + note.initializeJobListenerForParagraph(paragraph); } paragraph.run(); @@ -857,12 +843,14 @@ private void handleParagraphParams(String message, Note note, Paragraph paragrap throws IOException { // handle params if presented if (!StringUtils.isEmpty(message)) { - RunParagraphWithParametersRequest request = gson.fromJson(message, - RunParagraphWithParametersRequest.class); + RunParagraphWithParametersRequest request = + gson.fromJson(message, RunParagraphWithParametersRequest.class); Map paramsForUpdating = request.getParams(); if (paramsForUpdating != null) { paragraph.settings.getParams().putAll(paramsForUpdating); - note.persist(); + AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal()); + note.setLastReplName(paragraph.getId()); + note.persist(subject); } } } diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java index e2194fd6741..31df6181a2d 100644 --- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java +++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java @@ -17,6 +17,8 @@ package org.apache.zeppelin.notebook; +import static java.lang.String.format; + import java.io.IOException; import java.io.Serializable; import java.util.HashMap; @@ -172,6 +174,28 @@ void setInterpreterFactory(InterpreterFactory factory) { } } + public void initializeJobListenerForParagraph(Paragraph paragraph) { + final Note paragraphNote = paragraph.getNote(); + if (paragraphNote.getId().equals(this.getId())) { + throw new IllegalArgumentException(format("The paragraph %s from note %s " + + "does not belong to note %s", paragraph.getId(), paragraphNote.getId(), + this.getId())); + } + + boolean foundParagraph = false; + for (Paragraph ownParagraph : paragraphs) { + if (paragraph.getId().equals(ownParagraph.getId())) { + paragraph.setListener(this.jobListenerFactory.getParagraphJobListener(this)); + foundParagraph = true; + } + } + + if (!foundParagraph) { + throw new IllegalArgumentException(format("Cannot find paragraph %s " + + "from note %s", paragraph.getId(), paragraphNote.getId())); + } + } + void setJobListenerFactory(JobListenerFactory jobListenerFactory) { this.jobListenerFactory = jobListenerFactory; } @@ -484,7 +508,7 @@ public void run(String paragraphId) { logger.debug("New paragraph: {}", pText); p.setEffectiveText(pText); } else { - String intpExceptionMsg = String.format("%s", + String intpExceptionMsg = format("%s", p.getJobName() + "'s Interpreter " + requiredReplName + " not found"