[MINOR] Refresh search indexes after paragraph edits

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();
       });

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;
       });

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<String, Object> searchResponse = gson.fromJson(
+          EntityUtils.toString(search.getEntity(), StandardCharsets.UTF_8),
+          new TypeToken<Map<String, Object>>() {}.getType());
+      search.close();
+      List<Map<String, String>> searchResults =
+          (List<Map<String, String>>) searchResponse.get("body");
+      String expectedResultId = noteId + "/paragraph/" + newParagraphId;
+      assertTrue(searchResults.stream().anyMatch(result ->
+          result.get("id").startsWith(expectedResultId)));
     } finally {
       //cleanup
       if (null != noteId) {

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<Map<String, String>> searchResults = searchService.query(serviceSearchToken);
+    assertTrue(searchResults.stream().anyMatch(result ->
+        result.get("id").startsWith(note1Id)));
 
     // move paragraph
     reset(callback);