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 3c09a612f42..ff761eba6d0 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 @@ -706,6 +706,7 @@ public Response updateParagraph(@PathParam("noteId") String noteId, AuthenticationInfo subject = new AuthenticationInfo(user); notebook.saveNote(note, subject); + note.fireParagraphUpdateEvent(p); notebookServer.broadcastParagraph(note, p, MSG_ID_NOT_DEFINED); return new JsonResponse<>(Status.OK, "").build(); }); diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java b/zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java index eacd970eb05..1eb19a6decd 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java @@ -804,6 +804,7 @@ public void updateParagraph(String noteId, p.setText(text); } notebook.saveNote(note, context.getAutheInfo()); + note.fireParagraphUpdateEvent(p); callback.onSuccess(p, context); return null; }); diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinRestApiTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinRestApiTest.java index 385050ba77a..64321e9048b 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinRestApiTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinRestApiTest.java @@ -828,7 +828,7 @@ void testInsertParagraph() throws IOException { } @Test - void testUpdateParagraph() throws IOException { + void testUpdateParagraph() throws IOException, InterruptedException { String noteId = null; try { noteId = notebook.createNote("note1_testUpdateParagraph", anonymous); @@ -862,7 +862,9 @@ void testUpdateParagraph() throws IOException { return null; }); - String updateBothRequest = "{\"title\": \"updated title\", \"text\" : \"updated text 2\" }"; + String restSearchToken = "restSearchUpdatedToken"; + String updateBothRequest = "{\"title\": \"updated title\", \"text\" : \"" + + restSearchToken + "\" }"; CloseableHttpResponse updatePut = httpPut("/notebook/" + noteId + "/paragraph/" + newParagraphId, updateBothRequest); updatePut.close(); @@ -871,9 +873,22 @@ void testUpdateParagraph() throws IOException { noteP -> { Paragraph updatedBothParagraph = noteP.getParagraph(newParagraphId); assertEquals("updated title", updatedBothParagraph.getTitle()); - assertEquals("updated text 2", updatedBothParagraph.getText()); + assertEquals(restSearchToken, updatedBothParagraph.getText()); return null; }); + + // SearchService handles paragraph events asynchronously. + Thread.sleep(1000); + CloseableHttpResponse search = httpGet("/notebook/search?q=" + restSearchToken); + Map searchResponse = gson.fromJson( + EntityUtils.toString(search.getEntity(), StandardCharsets.UTF_8), + new TypeToken>() {}.getType()); + search.close(); + List> searchResults = + (List>) searchResponse.get("body"); + String expectedResultId = noteId + "/paragraph/" + newParagraphId; + assertTrue(searchResults.stream().anyMatch(result -> + result.get("id").startsWith(expectedResultId))); } finally { //cleanup if (null != noteId) { diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java index 1c4f554d50a..634a36cfed4 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java @@ -548,7 +548,7 @@ void testRenameNoteRejectsDuplicate() throws IOException { @Test - void testParagraphOperations() throws IOException { + void testParagraphOperations() throws IOException, InterruptedException { // create note String note1Id = notebookService.createNote("note1", "python", false, context, callback); notebook.processNote(note1Id, @@ -572,12 +572,21 @@ void testParagraphOperations() throws IOException { return null; }); - // update paragraph + // update paragraph and verify the asynchronous search listener sees the new text reset(callback); - notebookService.updateParagraph(note1Id, p.getId(), "my_title", "my_text", + String serviceSearchToken = "serviceSearchUpdatedToken"; + notebookService.updateParagraph(note1Id, p.getId(), "my_title", serviceSearchToken, new HashMap<>(), new HashMap<>(), context, callback); assertEquals("my_title", p.getTitle()); - assertEquals("my_text", p.getText()); + assertEquals(serviceSearchToken, p.getText()); + while (!searchService.isEventQueueEmpty()) { + Thread.sleep(10); + } + // The queue may be empty while its worker is finishing the current event. + Thread.sleep(100); + List> searchResults = searchService.query(serviceSearchToken); + assertTrue(searchResults.stream().anyMatch(result -> + result.get("id").startsWith(note1Id))); // move paragraph reset(callback);