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

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

@@ -229,7 +229,7 @@ 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());
     } else {
       appListener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), event.getIndex(),

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 {
 
@@ -70,7 +69,16 @@ public void run() {
     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;
+        listener.onOutputUpdated(update.getNoteId(), update.getParagraphId(), update.getIndex(),
+            update.getType(), update.getData());
+        continue;
+      }
+
       String noteId = buffer.getNoteId();
       String paragraphId = buffer.getParagraphId();
       int index = buffer.getIndex();
@@ -82,6 +90,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 +99,32 @@ public void run() {
       LOGGER.debug("Processing time for append-output took {} milliseconds", processingTime);
     }
 
-    Long sizeProcessed = Long.valueOf(0);
+    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();
       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);
-    }
+    stringBufferMap.clear();
+    return sizeProcessed;
   }
 
   public void appendBuffer(String noteId, String paragraphId, int index, String outputToAppend) {
     queue.offer(new AppendOutputBuffer(noteId, paragraphId, index, outputToAppend));
   }
 
+  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/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java

@@ -17,12 +17,14 @@
 
 package org.apache.zeppelin.interpreter.remote;
 
+import org.apache.zeppelin.interpreter.InterpreterResult;
 import org.apache.log4j.AppenderSkeleton;
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
 import org.apache.log4j.spi.LoggingEvent;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
@@ -39,6 +41,7 @@
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.Mockito.atMost;
 import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -88,6 +91,24 @@ public void testMultipleEventsOfSameParagraph() throws InterruptedException {
     verify(listener, times(1)).onOutputAppend(note1, para1, 0, "data1\ndata2\ndata3\n");
   }
 
+  @Test
+  void testUpdateDoesNotOvertakeQueuedAppend() {
+    RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class);
+    AppendOutputRunner runner = new AppendOutputRunner(listener);
+    runner.appendBuffer("note", "para", 0, "before-1\n");
+    runner.appendBuffer("note", "para", 0, "before-2\n");
+    runner.updateBuffer("note", "para", 0, InterpreterResult.Type.TEXT, "replacement\n");
+    runner.appendBuffer("note", "para", 0, "after\n");
+
+    runner.run();
+
+    InOrder order = inOrder(listener);
+    order.verify(listener).onOutputAppend("note", "para", 0, "before-1\nbefore-2\n");
+    order.verify(listener).onOutputUpdated(
+        "note", "para", 0, InterpreterResult.Type.TEXT, "replacement\n");
+    order.verify(listener).onOutputAppend("note", "para", 0, "after\n");
+  }
+
   @Test
   void testMultipleEventsOfDifferentParagraphs() throws InterruptedException {
     RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class);

zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts

@@ -15,10 +15,11 @@ import { NotebookParagraphPage } from 'e2e/models/notebook-paragraph-page';
 import { NotebookKeyboardPage } from 'e2e/models/notebook-keyboard-page';
 import {
   addPageAnnotationBeforeEach,
+  createTestNotebook,
   performLoginIfRequired,
-  waitForZeppelinReady,
   PAGES,
-  createTestNotebook
+  setParagraphText,
+  waitForZeppelinReady
 } from '../../../utils';
 
 test.describe('Notebook Paragraph Functionality', () => {
@@ -94,6 +95,32 @@ test.describe('Notebook Paragraph Functionality', () => {
     await expect(paragraphPage.resultDisplay).not.toBeEmpty();
   });
 
+  test('should accumulate interpreter output while the paragraph is running', async ({ page }) => {
+    await test.step('Given a shell paragraph that emits three delayed output chunks', async () => {
+      await setParagraphText(
+        page,
+        testNotebook.noteId,
+        testNotebook.paragraphId,
+        '%sh\necho first; sleep 3; echo second; sleep 5; echo third'
+      );
+      await page.reload();
+      await expect(paragraphPage.paragraphContainer).toBeVisible({ timeout: 30000 });
+    });
+
+    await test.step('When the paragraph runs', async () => {
+      await paragraphPage.runParagraph();
+    });
+
+    await test.step('Then output accumulates before the paragraph finishes', async () => {
+      await expect(paragraphPage.resultDisplay).toContainText('first', { timeout: 30000 });
+      await expect(paragraphPage.status).toHaveText('RUNNING');
+      await expect(paragraphPage.resultDisplay).toContainText(/first\s+second/, { timeout: 10000 });
+      await expect(paragraphPage.status).toHaveText('RUNNING');
+      await expect(paragraphPage.resultDisplay).toContainText(/first\s+second\s+third/, { timeout: 10000 });
+      await expect(paragraphPage.status).toHaveText('FINISHED');
+    });
+  });
+
   test('should display dynamic forms', async ({ page }) => {
     test.skip(!!process.env.CI, 'Dynamic form tests require a Spark interpreter — skipped on CI');
 

zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.spec.ts

@@ -0,0 +1,38 @@
+/*
+ * Licensed 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.
+ */
+
+import { expect, expectTypeOf, it } from 'vitest';
+
+import { MessageReceiveDataTypeMap } from './message-data-type-map.interface';
+import { OP } from './message-operator.interface';
+import { DatasetType, ParagraphAppendOutput, ParagraphUpdateOutput } from './message-paragraph.interface';
+
+it('declares the asymmetric paragraph output payloads sent by the server', () => {
+  const append: MessageReceiveDataTypeMap[OP.PARAGRAPH_APPEND_OUTPUT] = {
+    noteId: 'note',
+    paragraphId: 'paragraph',
+    index: 0,
+    data: 'chunk'
+  };
+  const update: MessageReceiveDataTypeMap[OP.PARAGRAPH_UPDATE_OUTPUT] = {
+    ...append,
+    type: DatasetType.TEXT
+  };
+  +expect(append).not.toHaveProperty('type');
+  expect(update.type).toBe(DatasetType.TEXT);
+  expectTypeOf<MessageReceiveDataTypeMap[OP.PARAGRAPH_APPEND_OUTPUT]>().toEqualTypeOf<ParagraphAppendOutput>();
+  expectTypeOf<MessageReceiveDataTypeMap[OP.PARAGRAPH_APPEND_OUTPUT]>().not.toHaveProperty('type');
+  expectTypeOf<MessageReceiveDataTypeMap[OP.PARAGRAPH_UPDATE_OUTPUT]>().toEqualTypeOf<ParagraphUpdateOutput>();
+  expectTypeOf<MessageReceiveDataTypeMap[OP.PARAGRAPH_UPDATE_OUTPUT]>()
+    .toHaveProperty('type')
+    .toEqualTypeOf<DatasetType>();
+});

zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.ts

@@ -67,12 +67,14 @@ import {
   CopyParagraph,
   InsertParagraph,
   MoveParagraph,
+  ParagraphAppendOutput,
   ParagraphClearAllOutput,
   ParagraphClearOutput,
   ParagraphExecutedBySpell,
   ParagraphRemove,
   ParagraphRemoved,
   ParagraphStatus,
+  ParagraphUpdateOutput,
   ParasInfo,
   PatchParagraphReceived,
   PatchParagraphSend,
@@ -108,6 +110,8 @@ export interface MessageReceiveDataTypeMap {
   [OP.IMPORT_NOTE]: ImportNoteReceived;
   [OP.SAVE_NOTE_FORMS]: SaveNoteFormsSend;
   [OP.PARAGRAPH]: UpdateParagraph;
+  [OP.PARAGRAPH_APPEND_OUTPUT]: ParagraphAppendOutput;
+  [OP.PARAGRAPH_UPDATE_OUTPUT]: ParagraphUpdateOutput;
   [OP.PATCH_PARAGRAPH]: PatchParagraphSend;
   [OP.PARAGRAPH_REMOVED]: ParagraphRemoved;
   [OP.EDITOR_SETTING]: EditorSettingReceived;

zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-paragraph.interface.ts

@@ -97,6 +97,17 @@ export class ParagraphIResultsMsgItem {
   data = '';
 }
 
+export interface ParagraphAppendOutput {
+  noteId: string;
+  paragraphId: string;
+  index: number;
+  data: string;
+}
+
+export interface ParagraphUpdateOutput extends ParagraphAppendOutput {
+  type: DatasetType;
+}
+
 export interface ParasInfo {
   id: string;
   infos: RuntimeInfos;