NotebookRestApi extends AbstractRestApi { @POST @ZeppelinApi public Response createNote(String message) throws IOException { String user = authenticationService.getPrincipal(); LOGGER.info("Creating new note by JSON {}", message); NewNoteRequest request = NewNoteRequest.fromJson(message); String defaultInterpreterGroup = request.getDefaultInterpreterGroup(); if (StringUtils.isBlank(defaultInterpreterGroup)) { defaultInterpreterGroup = zConf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_GROUP_DEFAULT); } Note note = notebookService.createNote( request.getName(), defaultInterpreterGroup, false, getServiceContext(), new RestServiceCallback<>()); AuthenticationInfo subject = new AuthenticationInfo(authenticationService.getPrincipal()); if (request.getParagraphs() != null) { for (NewParagraphRequest paragraphRequest : request.getParagraphs()) { Paragraph p = note.addNewParagraph(subject); initParagraph(p, paragraphRequest, user); } } return new JsonResponse(Status.OK, "", note.getId()).build(); } @Inject  NotebookRestApi(
      Notebook notebook,
      NotebookServer notebookServer,
      NotebookService notebookService,
      SearchService search,
      AuthorizationService authorizationService,
      ZeppelinConfiguration zConf,
      AuthenticationService authenticationService,
      JobManagerService jobManagerService,
      SchedulerService schedulerService); @GET @Path("{noteId}/permissions") @ZeppelinApi Response getNotePermissions(@PathParam("noteId") String noteId); @PUT @Path("{noteId}/permissions") @ZeppelinApi Response putNotePermissions(@PathParam("noteId") String noteId, String req); @GET @ZeppelinApi Response getNoteList(); @GET @Path("{noteId}") @ZeppelinApi Response getNote(@PathParam("noteId") String noteId); @GET @Path("export/{noteId}") @ZeppelinApi Response exportNote(@PathParam("noteId") String noteId); @POST @Path("import") @ZeppelinApi Response importNote(@QueryParam("notePath") String notePath, String noteJson); @POST @ZeppelinApi Response createNote(String message); @DELETE @Path("{noteId}") @ZeppelinApi Response deleteNote(@PathParam("noteId") String noteId); @POST @Path("{noteId}") @ZeppelinApi Response cloneNote(@PathParam("noteId") String noteId, String message); @PUT @Path("{noteId}/rename") @ZeppelinApi Response renameNote(@PathParam("noteId") String noteId,
                             String message); @POST @Path("{noteId}/paragraph") @ZeppelinApi Response insertParagraph(@PathParam("noteId") String noteId, String message); @GET @Path("{noteId}/paragraph/{paragraphId}") @ZeppelinApi Response getParagraph(@PathParam("noteId") String noteId,
                               @PathParam("paragraphId") String paragraphId); @PUT @Path("{noteId}/paragraph/{paragraphId}") @ZeppelinApi Response updateParagraph(@PathParam("noteId") String noteId,
                                  @PathParam("paragraphId") String paragraphId,
                                  String message); @PUT @Path("{noteId}/paragraph/{paragraphId}/config") @ZeppelinApi Response updateParagraphConfig(@PathParam("noteId") String noteId,
                                        @PathParam("paragraphId") String paragraphId,
                                        String message); @POST @Path("{noteId}/paragraph/{paragraphId}/move/{newIndex}") @ZeppelinApi Response moveParagraph(@PathParam("noteId") String noteId,
                                @PathParam("paragraphId") String paragraphId,
                                @PathParam("newIndex") String newIndex); @DELETE @Path("{noteId}/paragraph/{paragraphId}") @ZeppelinApi Response deleteParagraph(@PathParam("noteId") String noteId,
                                  @PathParam("paragraphId") String paragraphId); @POST @Path("{noteId}/paragraph/next") Response nextSessionParagraph(@PathParam("noteId") String noteId,
                                       @QueryParam("maxParagraph") int maxParagraph); @PUT @Path("{noteId}/clear") @ZeppelinApi Response clearAllParagraphOutput(@PathParam("noteId") String noteId); @POST @Path("job/{noteId}") @ZeppelinApi Response runNoteJobs(@PathParam("noteId") String noteId,
                              @QueryParam("blocking") Boolean blocking,
                              @QueryParam("isolated") Boolean isolated,
                              String message); @DELETE @Path("job/{noteId}") @ZeppelinApi Response stopNoteJobs(@PathParam("noteId") String noteId); @GET @Path("job/{noteId}") @ZeppelinApi Response getNoteJobStatus(@PathParam("noteId") String noteId); @GET @Path("job/{noteId}/{paragraphId}") @ZeppelinApi Response getNoteParagraphJobStatus(@PathParam("noteId") String noteId,
                                            @PathParam("paragraphId") String paragraphId); @POST @Path("job/{noteId}/{paragraphId}") @ZeppelinApi Response runParagraph(@PathParam("noteId") String noteId,
                               @PathParam("paragraphId") String paragraphId,
                               @QueryParam("sessionId") String sessionId,
                               String message); @POST @Path("run/{noteId}/{paragraphId}") @ZeppelinApi Response runParagraphSynchronously(@PathParam("noteId") String noteId,
                                            @PathParam("paragraphId") String paragraphId,
                                            @QueryParam("sessionId") String sessionId,
                                            String message); @DELETE @Path("job/{noteId}/{paragraphId}") @ZeppelinApi Response cancelParagraph(@PathParam("noteId") String noteId,
                                  @PathParam("paragraphId") String paragraphId); @POST @Path("cron/{noteId}") @ZeppelinApi Response registerCronJob(@PathParam("noteId") String noteId, String message); @DELETE @Path("cron/{noteId}") @ZeppelinApi Response removeCronJob(@PathParam("noteId") String noteId); @GET @Path("cron/{noteId}") @ZeppelinApi Response getCronJob(@PathParam("noteId") String noteId); @GET @Path("jobmanager/") @ZeppelinApi Response getJobListforNote(); @GET @Path("jobmanager/{lastUpdateUnixtime}/") @ZeppelinApi Response getUpdatedJobListforNote(@PathParam("lastUpdateUnixtime") long lastUpdateUnixTime); @GET @Path("search") @ZeppelinApi Response search(@QueryParam("q") String queryTerm);  }
@Test public void testRunParagraphJob() throws Exception { LOG.info("Running testRunParagraphJob"); Note note1 = null; try { note1 = TestUtils.getInstance(Notebook.class).createNote("note1", anonymous); note1.addNewParagraph(AuthenticationInfo.ANONYMOUS); Paragraph p = note1.addNewParagraph(AuthenticationInfo.ANONYMOUS); PostMethod post = httpPost("/notebook/job/" + note1.getId() + "/" + p.getId(), ""); assertThat(post, isAllowed()); Map<String, Object> resp = gson.fromJson(post.getResponseBodyAsString(), new TypeToken<Map<String, Object>>() {}.getType()); assertEquals(resp.get("status"), "OK"); post.releaseConnection(); p.waitUntilFinished(); assertEquals(p.getStatus(), Job.Status.FINISHED); p.setText("test"); post = httpPost("/notebook/job/" + note1.getId() + "/" + p.getId(), ""); assertThat(post, isAllowed()); resp = gson.fromJson(post.getResponseBodyAsString(), new TypeToken<Map<String, Object>>() {}.getType()); assertEquals(resp.get("status"), "OK"); post.releaseConnection(); p.waitUntilFinished(); assertNotEquals(p.getStatus(), Job.Status.FINISHED); } finally { if (null != note1) { TestUtils.getInstance(Notebook.class).removeNote(note1, anonymous); } } }
@Test public void testRunNoteBlocking() throws IOException { LOG.info("Running testRunNoteBlocking"); Note note1 = null; try { note1 = TestUtils.getInstance(Notebook.class).createNote("note1", anonymous); Paragraph p1 = note1.addNewParagraph(AuthenticationInfo.ANONYMOUS); Paragraph p2 = note1.addNewParagraph(AuthenticationInfo.ANONYMOUS); p1.setText("%python from __future__ import print_function\nimport time\ntime.sleep(1)\nuser='abc'"); p2.setText("%python print(user)"); PostMethod post = httpPost("/notebook/job/" + note1.getId() + "?blocking=true", ""); assertThat(post, isAllowed()); Map<String, Object> resp = gson.fromJson(post.getResponseBodyAsString(), new TypeToken<Map<String, Object>>() {}.getType()); assertEquals(resp.get("status"), "OK"); post.releaseConnection(); assertEquals(Job.Status.FINISHED, p1.getStatus()); assertEquals(Job.Status.FINISHED, p2.getStatus()); assertEquals("abc\n", p2.getReturn().message().get(0).getData()); } finally { if (null != note1) { TestUtils.getInstance(Notebook.class).removeNote(note1, anonymous); } } }
@Test public void testRunNoteNonBlocking() throws Exception { LOG.info("Running testRunNoteNonBlocking"); Note note1 = null; try { note1 = TestUtils.getInstance(Notebook.class).createNote("note1", anonymous); Paragraph p1 = note1.addNewParagraph(AuthenticationInfo.ANONYMOUS); Paragraph p2 = note1.addNewParagraph(AuthenticationInfo.ANONYMOUS); p1.setText("%python import time\ntime.sleep(5)\nname='hello'\nz.put('name', name)"); p2.setText("%sh(interpolate=true) echo '{name}'"); PostMethod post = httpPost("/notebook/job/" + note1.getId() + "?blocking=true", ""); assertThat(post, isAllowed()); Map<String, Object> resp = gson.fromJson(post.getResponseBodyAsString(), new TypeToken<Map<String, Object>>() {}.getType()); assertEquals(resp.get("status"), "OK"); post.releaseConnection(); p1.waitUntilFinished(); p2.waitUntilFinished(); assertEquals(Job.Status.FINISHED, p1.getStatus()); assertEquals(Job.Status.FINISHED, p2.getStatus()); assertEquals("hello\n", p2.getReturn().message().get(0).getData()); } finally { if (null != note1) { TestUtils.getInstance(Notebook.class).removeNote(note1, anonymous); } } }
@Test public void testRunNoteBlocking_Isolated() throws IOException { LOG.info("Running testRunNoteBlocking_Isolated"); Note note1 = null; try { InterpreterSettingManager interpreterSettingManager = TestUtils.getInstance(InterpreterSettingManager.class); InterpreterSetting interpreterSetting = interpreterSettingManager.getInterpreterSettingByName("python"); int pythonProcessNum = interpreterSetting.getAllInterpreterGroups().size(); note1 = TestUtils.getInstance(Notebook.class).createNote("note1", anonymous); Paragraph p1 = note1.addNewParagraph(AuthenticationInfo.ANONYMOUS); Paragraph p2 = note1.addNewParagraph(AuthenticationInfo.ANONYMOUS); p1.setText("%python from __future__ import print_function\nimport time\ntime.sleep(1)\nuser='abc'"); p2.setText("%python print(user)"); PostMethod post = httpPost("/notebook/job/" + note1.getId() + "?blocking=true&isolated=true", ""); assertThat(post, isAllowed()); Map<String, Object> resp = gson.fromJson(post.getResponseBodyAsString(), new TypeToken<Map<String, Object>>() {}.getType()); assertEquals(resp.get("status"), "OK"); post.releaseConnection(); assertEquals(Job.Status.FINISHED, p1.getStatus()); assertEquals(Job.Status.FINISHED, p2.getStatus()); assertEquals("abc\n", p2.getReturn().message().get(0).getData()); assertEquals(pythonProcessNum, interpreterSetting.getAllInterpreterGroups().size()); } finally { if (null != note1) { TestUtils.getInstance(Notebook.class).removeNote(note1, anonymous); } } }
@Test public void testRunNoteNonBlocking_Isolated() throws IOException, InterruptedException { LOG.info("Running testRunNoteNonBlocking_Isolated"); Note note1 = null; try { InterpreterSettingManager interpreterSettingManager = TestUtils.getInstance(InterpreterSettingManager.class); InterpreterSetting interpreterSetting = interpreterSettingManager.getInterpreterSettingByName("python"); int pythonProcessNum = interpreterSetting.getAllInterpreterGroups().size(); note1 = TestUtils.getInstance(Notebook.class).createNote("note1", anonymous); Paragraph p1 = note1.addNewParagraph(AuthenticationInfo.ANONYMOUS); Paragraph p2 = note1.addNewParagraph(AuthenticationInfo.ANONYMOUS); p1.setText("%python from __future__ import print_function\nimport time\ntime.sleep(1)\nuser='abc'"); p2.setText("%python print(user)"); PostMethod post = httpPost("/notebook/job/" + note1.getId() + "?blocking=false&isolated=true", ""); assertThat(post, isAllowed()); Map<String, Object> resp = gson.fromJson(post.getResponseBodyAsString(), new TypeToken<Map<String, Object>>() {}.getType()); assertEquals(resp.get("status"), "OK"); post.releaseConnection(); while(note1.isRunning()) { Thread.sleep(1000); } assertEquals(Job.Status.FINISHED, p1.getStatus()); assertEquals(Job.Status.FINISHED, p2.getStatus()); assertEquals("abc\n", p2.getReturn().message().get(0).getData()); assertEquals(pythonProcessNum, interpreterSetting.getAllInterpreterGroups().size()); } finally { if (null != note1) { TestUtils.getInstance(Notebook.class).removeNote(note1, anonymous); } } }
@Test public void testRunNoteWithParams() throws IOException, InterruptedException { Note note1 = null; try { note1 = TestUtils.getInstance(Notebook.class).createNote("note1", anonymous); Paragraph p1 = note1.addNewParagraph(AuthenticationInfo.ANONYMOUS); Paragraph p2 = note1.addNewParagraph(AuthenticationInfo.ANONYMOUS); p1.setText("%python name = z.input('name', 'world')\nprint(name)"); p2.setText("%sh echo '${name=world}'"); Map<String, Object> paramsMap = new HashMap<>(); paramsMap.put("name", "zeppelin"); ParametersRequest parametersRequest = new ParametersRequest(paramsMap); PostMethod post = httpPost("/notebook/job/" + note1.getId() + "?blocking=false&isolated=true&", parametersRequest.toJson()); assertThat(post, isAllowed()); Map<String, Object> resp = gson.fromJson(post.getResponseBodyAsString(), new TypeToken<Map<String, Object>>() {}.getType()); assertEquals(resp.get("status"), "OK"); post.releaseConnection(); while(note1.isRunning()) { Thread.sleep(1000); } assertEquals(Job.Status.FINISHED, p1.getStatus()); assertEquals(Job.Status.FINISHED, p2.getStatus()); assertEquals("zeppelin\n", p1.getReturn().message().get(0).getData()); assertEquals("zeppelin\n", p2.getReturn().message().get(0).getData()); post = httpPost("/notebook/job/" + note1.getId() + "?blocking=false&isolated=true", ""); assertThat(post, isAllowed()); resp = gson.fromJson(post.getResponseBodyAsString(), new TypeToken<Map<String, Object>>() {}.getType()); assertEquals(resp.get("status"), "OK"); post.releaseConnection(); while(note1.isRunning()) { Thread.sleep(1000); } assertEquals(Job.Status.FINISHED, p1.getStatus()); assertEquals(Job.Status.FINISHED, p2.getStatus()); assertEquals("world\n", p1.getReturn().message().get(0).getData()); assertEquals("world\n", p2.getReturn().message().get(0).getData()); } finally { if (null != note1) { TestUtils.getInstance(Notebook.class).removeNote(note1, anonymous); } } }
@Test public void testRunAllParagraph_FirstFailed() throws IOException { LOG.info("Running testRunAllParagraph_FirstFailed"); Note note1 = null; try { note1 = TestUtils.getInstance(Notebook.class).createNote("note1", anonymous); Paragraph p1 = note1.addNewParagraph(AuthenticationInfo.ANONYMOUS); Paragraph p2 = note1.addNewParagraph(AuthenticationInfo.ANONYMOUS); p1.setText("%python from __future__ import print_function\nimport time\ntime.sleep(1)\nprint(user2)"); p2.setText("%python user2='abc'\nprint(user2)"); PostMethod post = httpPost("/notebook/job/" + note1.getId() + "?blocking=true", ""); assertThat(post, isAllowed()); assertEquals(Job.Status.ERROR, p1.getStatus()); assertEquals(Job.Status.READY, p2.getStatus()); } finally { if (null != note1) { TestUtils.getInstance(Notebook.class).removeNote(note1, anonymous); } } }