[ZEPPELIN-6659] Render streaming interpreter output in Angular New UI

zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java

@@ -229,8 +229,10 @@ public void appendOutput(OutputAppendEvent event) throws InterpreterRPCException
   @Override
   public void updateOutput(OutputUpdateEvent event) throws InterpreterRPCException, TException {
     if (event.getAppId() == null) {
-      listener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), event.getIndex(),
+      runner.updateBuffer(event.getNoteId(), event.getParagraphId(), event.getIndex(),
           InterpreterResult.Type.valueOf(event.getType()), event.getData());
+      // Complete replacements before the interpreter can publish its terminal result.
+      runner.run();
     } else {
       appListener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), event.getIndex(),
           event.getAppId(), InterpreterResult.Type.valueOf(event.getType()), event.getData());
@@ -239,11 +241,15 @@ public void updateOutput(OutputUpdateEvent event) throws InterpreterRPCException
 
   @Override
   public void updateAllOutput(OutputUpdateAllEvent event) throws InterpreterRPCException, TException {
-    listener.onOutputClear(event.getNoteId(), event.getParagraphId());
-    for (int i = 0; i < event.getMsg().size(); i++) {
-      RemoteInterpreterResultMessage msg = event.getMsg().get(i);
-      listener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), i,
-          InterpreterResult.Type.valueOf(msg.getType()), msg.getData());
+    synchronized (runner) {
+      // Finish earlier output before the clear; keep replacements ahead of the next drain.
+      runner.run();
+      listener.onOutputClear(event.getNoteId(), event.getParagraphId());
+      for (int i = 0; i < event.getMsg().size(); i++) {
+        RemoteInterpreterResultMessage msg = event.getMsg().get(i);
+        listener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), i,
+            InterpreterResult.Type.valueOf(msg.getType()), msg.getData());
+      }
     }
   }
 
@@ -266,6 +272,9 @@ public void updateAppStatus(AppStatusUpdateEvent event) throws InterpreterRPCExc
 
   @Override
   public void checkpointOutput(String noteId, String paragraphId) throws InterpreterRPCException, TException {
+    // Drain replacements before checkpointing.
+    // Keep storage callbacks outside the runner lock to avoid blocking output delivery.
+    runner.run();
     listener.checkpointOutput(noteId, paragraphId);
   }
 

zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java

@@ -17,6 +17,7 @@
 
 package org.apache.zeppelin.interpreter.remote;
 
+import org.apache.zeppelin.interpreter.InterpreterResult;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -29,10 +30,8 @@
 import java.util.concurrent.LinkedBlockingQueue;
 
 /**
- * This thread sends paragraph's append-data
- * periodically, rather than continously, with
- * a period of BUFFER_TIME_MS. It handles append-data
- * for all paragraphs across all notebooks.
+ * Sends paragraph output periodically. Adjacent append events are batched, while update events
+ * share the same queue so that they cannot overtake earlier appends.
  */
 public class AppendOutputRunner implements Runnable {
 
@@ -48,29 +47,36 @@ public AppendOutputRunner(RemoteInterpreterProcessListener listener) {
     this.listener = listener;
   }
 
+  // Serialize scheduled and RPC drains to preserve callback order.
+  // Empty drains must return immediately to RPC callers.
   @Override
-  public void run() {
+  public synchronized void run() {
 
     Map<String, StringBuilder> stringBufferMap = new HashMap<>();
     List<AppendOutputBuffer> list = new LinkedList<>();
 
-    /* "drainTo" method does not wait for any element
-     * to be present in the queue, and thus this loop would
-     * continuosly run (with period of BUFFER_TIME_MS). "take()" method
-     * waits for the queue to become non-empty and then removes
-     * one element from it. Rest elements from queue (if present) are
-     * removed using "drainTo" method. Thus we save on some un-necessary
-     * cpu-cycles.
-     */
-    try {
-      list.add(queue.take());
-    } catch (InterruptedException e) {
-      LOGGER.error("Wait for OutputBuffer queue interrupted: {}", e.getMessage());
+    queue.drainTo(list);
+    if (list.isEmpty()) {
+      return;
     }
     Long processingStartTime = System.currentTimeMillis();
-    queue.drainTo(list);
 
-    for (AppendOutputBuffer buffer: list) {
+    Long sizeProcessed = Long.valueOf(0);
+    for (AppendOutputBuffer buffer : list) {
+      if (buffer instanceof UpdateOutputBuffer) {
+        sizeProcessed += flushAppendBuffers(stringBufferMap);
+        UpdateOutputBuffer update = (UpdateOutputBuffer) buffer;
+        try {
+          listener.onOutputUpdated(update.getNoteId(), update.getParagraphId(), update.getIndex(),
+              update.getType(), update.getData());
+        } catch (RuntimeException e) {
+          // A stale callback must not abort another paragraph's synchronous drain.
+          LOGGER.warn("Failed to update output for note {} paragraph {}",
+              update.getNoteId(), update.getParagraphId(), e);
+        }
+        continue;
+      }
+
       String noteId = buffer.getNoteId();
       String paragraphId = buffer.getParagraphId();
       int index = buffer.getIndex();
@@ -82,6 +88,7 @@ public void run() {
       builder.append(buffer.getData());
       stringBufferMap.put(stringBufferKey, builder);
     }
+    sizeProcessed += flushAppendBuffers(stringBufferMap);
     Long processingTime = System.currentTimeMillis() - processingStartTime;
 
     if (processingTime > SAFE_PROCESSING_TIME) {
@@ -90,24 +97,38 @@ public void run() {
       LOGGER.debug("Processing time for append-output took {} milliseconds", processingTime);
     }
 
-    Long sizeProcessed = Long.valueOf(0);
-    for (Entry<String, StringBuilder> stringBufferMapEntry : stringBufferMap.entrySet()) {
-      String stringBufferKey = stringBufferMapEntry.getKey();
-      StringBuilder buffer = stringBufferMapEntry.getValue();
-      sizeProcessed += buffer.length();
-      String[] keys = stringBufferKey.split(":");
-      listener.onOutputAppend(keys[0], keys[1], Integer.parseInt(keys[2]), buffer.toString());
-    }
-
     if (sizeProcessed > SAFE_PROCESSING_STRING_SIZE) {
       LOGGER.warn("Processing size for buffered append-output is high: {} characters.", sizeProcessed);
     } else {
       LOGGER.debug("Processing size for append-output is {} characters", sizeProcessed);
     }
   }
 
+  private long flushAppendBuffers(Map<String, StringBuilder> stringBufferMap) {
+    long sizeProcessed = 0;
+    for (Entry<String, StringBuilder> stringBufferMapEntry : stringBufferMap.entrySet()) {
+      String stringBufferKey = stringBufferMapEntry.getKey();
+      StringBuilder buffer = stringBufferMapEntry.getValue();
+      sizeProcessed += buffer.length();
+      try {
+        String[] keys = stringBufferKey.split(":");
+        listener.onOutputAppend(keys[0], keys[1], Integer.parseInt(keys[2]), buffer.toString());
+      } catch (RuntimeException e) {
+        // One stale append must not abort another paragraph's synchronous drain.
+        LOGGER.warn("Failed to append output for {}", stringBufferKey, e);
+      }
+    }
+    stringBufferMap.clear();
+    return sizeProcessed;
+  }
+
   public void appendBuffer(String noteId, String paragraphId, int index, String outputToAppend) {
     queue.offer(new AppendOutputBuffer(noteId, paragraphId, index, outputToAppend));
   }
 
+  /** Enqueues a replacement; callers needing completion must also invoke run(). */
+  public void updateBuffer(String noteId, String paragraphId, int index,
+                           InterpreterResult.Type type, String output) {
+    queue.offer(new UpdateOutputBuffer(noteId, paragraphId, index, type, output));
+  }
 }

zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/UpdateOutputBuffer.java

@@ -0,0 +1,41 @@
+/*
+ * 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.interpreter.remote;
+
+import org.apache.zeppelin.interpreter.InterpreterResult;
+
+/**
+ * This element stores the buffered update-data of paragraph's output. It shares the
+ * append-data queue so that an update, which replaces a result, can never be sent
+ * ahead of the appends that preceded it.
+ */
+public class UpdateOutputBuffer extends AppendOutputBuffer {
+
+  private final InterpreterResult.Type type;
+
+  public UpdateOutputBuffer(String noteId, String paragraphId, int index,
+                            InterpreterResult.Type type, String data) {
+    super(noteId, paragraphId, index, data);
+    this.type = type;
+  }
+
+  public InterpreterResult.Type getType() {
+    return type;
+  }
+
+}

zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java

@@ -1769,7 +1769,19 @@ public void onOutputAppend(String noteId, String paragraphId, int index, String
         .put("paragraphId", paragraphId)
         .put("index", index)
         .put("data", output);
-    connectionManager.broadcast(noteId, msg);
+    try {
+      getNotebook().processNote(noteId, note -> {
+        if (note == null) {
+          LOGGER.warn("Note {} not found", noteId);
+        } else if (!note.isPersonalizedMode()) {
+          // Streaming events do not identify the user that owns the execution.
+          connectionManager.broadcast(noteId, msg);
+        }
+        return null;
+      });
+    } catch (IOException e) {
+      LOGGER.warn("Fail to call onOutputAppend", e);
+    }
   }
 
   /**
@@ -1796,16 +1808,15 @@ public void onOutputUpdated(String noteId, String paragraphId, int index,
             LOGGER.warn("Note {} not found", noteId);
             return null;
           }
-          Paragraph paragraph = note.getParagraph(paragraphId);
-          paragraph.updateOutputBuffer(index, type, output);
           if (note.isPersonalizedMode()) {
-            String user = note.getParagraph(paragraphId).getUser();
-            if (null != user) {
-              connectionManager.multicastToUser(user, msg);
-            }
-          } else {
-            connectionManager.broadcast(noteId, msg);
+            // Streaming events carry no owner. The shared outputBuffer is what checkpointOutput
+            // saves as the shared result and what other users' paragraphs are cloned from, so
+            // one user's output must not be written there. Personalized clients get their
+            // user-specific terminal snapshot instead.
+            return null;
           }
+          note.getParagraph(paragraphId).updateOutputBuffer(index, type, output);
+          connectionManager.broadcast(noteId, msg);
           return null;
         });
     } catch (IOException e) {
@@ -1828,11 +1839,15 @@ public void onOutputClear(String noteId, String paragraphId) {
           if (note == null) {
             // It is possible the note is removed, but the job is still running
             LOGGER.warn("Note {} doesn't existed, it maybe deleted.", noteId);
-          } else {
-            note.clearParagraphOutput(paragraphId);
-            Paragraph paragraph = note.getParagraph(paragraphId);
-            broadcastParagraph(note, paragraph, MSG_ID_NOT_DEFINED);
+            return null;
           }
+          if (note.isPersonalizedMode()) {
+            // Streaming events carry no owner, so they must not mutate shared paragraph state.
+            return null;
+          }
+          note.clearParagraphOutput(paragraphId);
+          Paragraph paragraph = note.getParagraph(paragraphId);
+          broadcastParagraph(note, paragraph, MSG_ID_NOT_DEFINED);
           return null;
         });