This is an automated email from the ASF dual-hosted git repository.
voidmatcha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/zeppelin.git
The following commit(s) were added to refs/heads/master by this push:
new 0c35c61f40 [ZEPPELIN-6659] Render streaming interpreter output in
Angular New UI
0c35c61f40 is described below
commit 0c35c61f407c0d6ca6d237b261e795f21c83c524
Author: Minho Jang <[email protected]>
AuthorDate: Sun Sep 13 19:31:35 2026 +0900
[ZEPPELIN-6659] Render streaming interpreter output in Angular New UI
### What is this PR for?
The Angular New UI declares `PARAGRAPH_APPEND_OUTPUT` and
`PARAGRAPH_UPDATE_OUTPUT` in its WebSocket operation enum but does not consume
either message. Interpreter output emitted while a paragraph is running
is therefore discarded until the terminal `PARAGRAPH` snapshot arrives.
This PR adds the missing consumer based on the server's actual wire
contract. It declares the receive payloads in the SDK, handles both operations
in `ParagraphBase`, and introduces `ParagraphOutputState` to fold UPDATE,
APPEND, and terminal snapshots into the current paragraph result. APPEND data
is accumulated by result index, APPEND received before a typed UPDATE is held
until its type is known, and the terminal `PARAGRAPH` snapshot becomes
authoritative so late frames cannot dupli [...]
The PR also fixes a server-side ordering ambiguity. APPEND events were
buffered for up to 100 ms by `AppendOutputRunner`, while UPDATE events bypassed
that queue and could overtake an earlier APPEND. A client
cannot distinguish that delayed APPEND from one genuinely produced after
the UPDATE. APPEND and UPDATE now share the same queue, with UPDATE acting as
an ordering boundary: preceding APPEND chunks are flushed
before the UPDATE, and subsequent APPEND chunks remain after it.
A focused, versioned capture records the callback-order evidence and
observed WebSocket sequences with paragraph streaming enabled and disabled.
Reducer tests replay the capture and cover delayed-frame permutations. The
client remains defensive against late frames for compatibility with older
servers.
The change is limited to the standard paragraph result-rendering path.
Application output rendering is unchanged.
### What type of PR is it?
Bug Fix
### Todos
* [x] Declare the APPEND and UPDATE receive payloads and their type
asymmetry
* [x] Consume streaming paragraph output in the Angular New UI
* [x] Preserve APPEND and UPDATE ordering on the server
* [x] Cover coalesced, reordered, and late output events
* [x] Record and replay enabled and disabled streaming captures
* [x] Add browser coverage for output accumulation while RUNNING
* [x] Add the post-fix streaming scenario to the notebook parity registry
after apache/zeppelin#5456 lands
### What is the Jira issue?
* https://issues.apache.org/jira/browse/ZEPPELIN-6659
The reusable WebSocket fixture infrastructure tracked by ZEPPELIN-6665 and
ZEPPELIN-6671 is not available yet, so this PR includes the focused capture and
ordering replay required by ZEPPELIN-6659.
The notebook parity registry is being introduced by ZEPPELIN-6660 in
apache/zeppelin#5456. After that PR lands, this change will register streaming
output as a separate result scenario. The scenario will require the first chunk
to be visible while the paragraph is RUNNING, later chunks to accumulate rather
than replace earlier output, and the terminal result to contain every chunk
exactly once.
### How should this be tested?
* Run the server ordering regression test:
```bash
./mvnw test -pl zeppelin-server -Dtest=AppendOutputRunnerTest
```
* Run the frontend reducer and SDK contract tests:
```bash
cd zeppelin-web-angular
npm run test:shell --
src/app/core/paragraph-base/paragraph-output-state.spec.ts
projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.spec.ts
```
* Validate the WebSocket operation contract:
```bash
npm run check:websocket-contract
```
The browser test runs a shell paragraph that prints three chunks separated
by 3 and 5 seconds. It verifies that `first` is visible while the paragraph is
RUNNING, `second` is appended while the paragraph
remains RUNNING, and the FINISHED result contains `first`, `second`, and
`third`.
The capture replay additionally verifies coalesced APPEND chunks, APPEND
before a typed UPDATE, UPDATE overtaking a queued APPEND, APPEND after the
terminal `PARAGRAPH`, and terminal fallback when
`zeppelin.websocket.paragraph_status_progress.enable=false`.
### Questions:
* Does the license files need to update? No.
* Is there breaking changes for older versions? No.
* Does this needs documentation? No.
Closes #5460 from miinhho/fix/streaming-interpreter-output.
Signed-off-by: YONGJAE LEE <[email protected]>
---
.../interpreter/RemoteInterpreterEventServer.java | 21 +-
.../interpreter/remote/AppendOutputRunner.java | 77 ++++---
.../interpreter/remote/UpdateOutputBuffer.java | 41 ++++
.../org/apache/zeppelin/socket/NotebookServer.java | 41 ++--
.../RemoteInterpreterEventServerTest.java | 128 +++++++++++
.../interpreter/remote/AppendOutputRunnerTest.java | 128 +++++++++--
.../socket/NotebookServerStreamingScopeTest.java | 131 +++++++++++
.../e2e/models/published-paragraph-page.ts | 2 +
.../e2e/scenarios/notebook-parity.json | 80 +++++++
.../e2e/scenarios/notebook-parity.md | 13 ++
.../paragraph/paragraph-functionality.spec.ts | 36 ++-
.../notebook/published/published-streaming.spec.ts | 120 ++++++++++
.../message-data-type-map.interface.spec.ts | 38 ++++
.../interfaces/message-data-type-map.interface.ts | 4 +
.../src/interfaces/message-paragraph.interface.ts | 11 +
.../app/core/paragraph-base/paragraph-base.spec.ts | 247 +++++++++++++++++++++
.../src/app/core/paragraph-base/paragraph-base.ts | 94 +++++++-
.../paragraph-base/paragraph-output-state.spec.ts | 151 +++++++++++++
.../core/paragraph-base/paragraph-output-state.ts | 79 +++++++
.../paragraph-output-stream.capture.json | 148 ++++++++++++
.../notebook/paragraph/paragraph.component.ts | 9 +-
.../paragraph/paragraph.component.spec.ts | 143 ++++++++++++
.../published/paragraph/paragraph.component.ts | 18 +-
23 files changed, 1681 insertions(+), 79 deletions(-)
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java
index 657ad593c8..c8f8886e6a 100644
---
a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java
+++
b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java
@@ -229,8 +229,10 @@ public class RemoteInterpreterEventServer implements
RemoteInterpreterEventServi
@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 class RemoteInterpreterEventServer implements
RemoteInterpreterEventServi
@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 class RemoteInterpreterEventServer implements
RemoteInterpreterEventServi
@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);
}
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java
index 93dd328296..8bc063af10 100644
---
a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java
+++
b/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.BlockingQueue;
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 class AppendOutputRunner implements Runnable {
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 class AppendOutputRunner implements Runnable {
builder.append(buffer.getData());
stringBufferMap.put(stringBufferKey, builder);
}
+ sizeProcessed += flushAppendBuffers(stringBufferMap);
Long processingTime = System.currentTimeMillis() - processingStartTime;
if (processingTime > SAFE_PROCESSING_TIME) {
@@ -90,15 +97,6 @@ public class AppendOutputRunner implements Runnable {
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 {
@@ -106,8 +104,31 @@ public class AppendOutputRunner implements Runnable {
}
}
+ 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));
+ }
}
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/UpdateOutputBuffer.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/UpdateOutputBuffer.java
new file mode 100644
index 0000000000..15de2d5092
--- /dev/null
+++
b/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;
+ }
+
+}
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
index 000c565eea..4d31a06558 100644
---
a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
+++
b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
@@ -1767,7 +1767,19 @@ public class NotebookServer implements
AngularObjectRegistryListener,
.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);
+ }
}
/**
@@ -1794,16 +1806,15 @@ public class NotebookServer implements
AngularObjectRegistryListener,
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) {
@@ -1826,11 +1837,15 @@ public class NotebookServer implements
AngularObjectRegistryListener,
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;
});
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java
index ad385c612e..db73b10a42 100644
---
a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java
@@ -20,24 +20,152 @@ import static
org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.Serializable;
+import java.lang.reflect.Field;
import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.remote.AppendOutputRunner;
import org.apache.zeppelin.interpreter.remote.InvokeResourceMethodEventMessage;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcessListener;
import org.apache.zeppelin.interpreter.thrift.InterpreterRPCException;
+import org.apache.zeppelin.interpreter.thrift.OutputAppendEvent;
+import org.apache.zeppelin.interpreter.thrift.OutputUpdateAllEvent;
+import org.apache.zeppelin.interpreter.thrift.OutputUpdateEvent;
+import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterResultMessage;
import org.apache.zeppelin.resource.Resource;
import org.apache.zeppelin.resource.ResourceId;
import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
+
public class RemoteInterpreterEventServerTest {
+ @Test
+ void updateOutputCompletesBeforeReturning() throws Exception {
+ RemoteInterpreterProcessListener listener =
mock(RemoteInterpreterProcessListener.class);
+ RemoteInterpreterEventServer server =
+ serverWithRunner(listener, new AppendOutputRunner(listener));
+ try {
+ server.updateOutput(new OutputUpdateEvent("note", "para", 0, "TEXT",
"final", null));
+ // A caller may publish terminal status as soon as the RPC returns.
+ verify(listener).onOutputUpdated("note", "para", 0,
InterpreterResult.Type.TEXT, "final");
+ } finally {
+ server.stop();
+ }
+ }
+
+ @Test
+ void checkpointDrainsPendingOutputBeforeSaving() throws Exception {
+ RemoteInterpreterProcessListener listener =
mock(RemoteInterpreterProcessListener.class);
+ RemoteInterpreterEventServer server =
+ serverWithRunner(listener, new AppendOutputRunner(listener));
+ try {
+ server.appendOutput(new OutputAppendEvent("note", "para", 0, "pending",
null));
+ server.checkpointOutput("note", "para");
+ InOrder order = inOrder(listener);
+ order.verify(listener).onOutputAppend("note", "para", 0, "pending");
+ order.verify(listener).checkpointOutput("note", "para");
+ } finally {
+ server.stop();
+ }
+ }
+
+ @Test
+ void updateAllIsAnOrderedClearAndReplacement() throws Exception {
+ RemoteInterpreterProcessListener listener =
mock(RemoteInterpreterProcessListener.class);
+ AppendOutputRunner runner = new AppendOutputRunner(listener);
+ RemoteInterpreterEventServer server = serverWithRunner(listener, runner);
+ try {
+ runner.appendBuffer("note", "para", 0, "old");
+ server.updateAllOutput(new OutputUpdateAllEvent("note", "para",
Collections.singletonList(
+ new RemoteInterpreterResultMessage("HTML", "replacement"))));
+ verify(listener).onOutputUpdated("note", "para", 0,
+ InterpreterResult.Type.HTML, "replacement");
+ runner.appendBuffer("note", "para", 0, "new");
+ runner.run();
+ InOrder order = inOrder(listener);
+ order.verify(listener).onOutputAppend("note", "para", 0, "old");
+ order.verify(listener).onOutputClear("note", "para");
+ order.verify(listener).onOutputUpdated("note", "para", 0,
+ InterpreterResult.Type.HTML, "replacement");
+ order.verify(listener).onOutputAppend("note", "para", 0, "new");
+ } finally {
+ server.stop();
+ }
+ }
+
+ @Test
+ void updateAllWaitsForInFlightAppendAndCompletesBeforeReturning() throws
Exception {
+ RemoteInterpreterProcessListener listener =
mock(RemoteInterpreterProcessListener.class);
+ AppendOutputRunner runner = new AppendOutputRunner(listener);
+ RemoteInterpreterEventServer server = serverWithRunner(listener, runner);
+ CountDownLatch entered = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ CountDownLatch updateStarted = new CountDownLatch(1);
+ doAnswer(invocation -> {
+ entered.countDown();
+ assertTrue(release.await(5, TimeUnit.SECONDS));
+ return null;
+ }).when(listener).onOutputAppend("note", "para", 0, "old");
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ runner.appendBuffer("note", "para", 0, "old");
+ Future<?> first = executor.submit(runner);
+ assertTrue(entered.await(5, TimeUnit.SECONDS));
+ Future<?> update = executor.submit(() -> {
+ updateStarted.countDown();
+ server.updateAllOutput(new OutputUpdateAllEvent("note", "para",
Collections.singletonList(
+ new RemoteInterpreterResultMessage("HTML", "replacement"))));
+ verify(listener).onOutputUpdated("note", "para", 0,
+ InterpreterResult.Type.HTML, "replacement");
+ return null;
+ });
+ assertTrue(updateStarted.await(5, TimeUnit.SECONDS));
+ assertThrows(TimeoutException.class, () -> update.get(100,
TimeUnit.MILLISECONDS));
+ release.countDown();
+ first.get(5, TimeUnit.SECONDS);
+ update.get(5, TimeUnit.SECONDS);
+ InOrder order = inOrder(listener);
+ order.verify(listener).onOutputAppend("note", "para", 0, "old");
+ order.verify(listener).onOutputClear("note", "para");
+ order.verify(listener).onOutputUpdated("note", "para", 0,
+ InterpreterResult.Type.HTML, "replacement");
+ } finally {
+ release.countDown();
+ executor.shutdownNow();
+ server.stop();
+ }
+ }
+
+ private RemoteInterpreterEventServer serverWithRunner(
+ RemoteInterpreterProcessListener listener, AppendOutputRunner runner)
throws Exception {
+ InterpreterSettingManager manager = mock(InterpreterSettingManager.class);
+ when(manager.getRemoteInterpreterProcessListener()).thenReturn(listener);
+ RemoteInterpreterEventServer server = new RemoteInterpreterEventServer(
+ mock(ZeppelinConfiguration.class), manager);
+ Field field =
RemoteInterpreterEventServer.class.getDeclaredField("runner");
+ field.setAccessible(true);
+ field.set(server, runner);
+ return server;
+ }
+
@Test
void invokeMethodThrowsRpcExceptionWhenSerializationFails() throws Exception
{
ZeppelinConfiguration zConf = mock(ZeppelinConfiguration.class);
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java
index 1d8d273cb7..2d6e08beba 100644
---
a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java
@@ -17,28 +17,39 @@
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;
import java.util.ArrayList;
import java.util.List;
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
-import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.doThrow;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
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 +99,24 @@ class AppendOutputRunnerTest {
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);
@@ -145,24 +174,85 @@ class AppendOutputRunnerTest {
logger.addAppender(appender);
runner.run();
- List<LoggingEvent> log;
-
- int warnLogCounter;
- LoggingEvent sizeWarnLogEntry = null;
- do {
- warnLogCounter = 0;
- log = appender.getLog();
- for (LoggingEvent logEntry: log) {
- if (Level.WARN.equals(logEntry.getLevel())) {
- sizeWarnLogEntry = logEntry;
- warnLogCounter += 1;
- }
- }
- } while(warnLogCounter != 2);
+ try {
+ String expected = "Processing size for buffered append-output is high: "
+ + (data.length() * numEvents) + " characters.";
+ assertTrue(appender.getLog().stream().anyMatch(event ->
+ Level.WARN.equals(event.getLevel()) &&
expected.equals(event.getMessage())));
+ } finally {
+ logger.removeAppender(appender);
+ }
+ }
+
+ @Test
+ void emptyDrainDoesNotBlock() {
+ AppendOutputRunner runner =
+ new AppendOutputRunner(mock(RemoteInterpreterProcessListener.class));
+ assertTimeoutPreemptively(Duration.ofSeconds(1), runner::run);
+ }
+
+ @Test
+ void updateFailureDoesNotDiscardOtherEventsOrLaterDrains() {
+ RemoteInterpreterProcessListener listener =
mock(RemoteInterpreterProcessListener.class);
+ AppendOutputRunner runner = new AppendOutputRunner(listener);
+ doThrow(new IllegalStateException("removed")).when(listener)
+ .onOutputUpdated("note", "gone", 0, InterpreterResult.Type.TEXT,
"bad");
+ runner.appendBuffer("note", "gone", 0, "bad");
+ runner.updateBuffer("note", "gone", 0, InterpreterResult.Type.TEXT, "bad");
+ runner.appendBuffer("note", "present", 0, "good");
+ runner.run();
+ runner.appendBuffer("note", "present", 0, "later");
+ runner.run();
+ verify(listener).onOutputAppend("note", "present", 0, "good");
+ verify(listener).onOutputAppend("note", "present", 0, "later");
+ }
+
+ @Test
+ void appendFailureDoesNotDiscardLaterUpdateOrDrain() {
+ RemoteInterpreterProcessListener listener =
mock(RemoteInterpreterProcessListener.class);
+ AppendOutputRunner runner = new AppendOutputRunner(listener);
+ doThrow(new IllegalStateException("removed")).when(listener)
+ .onOutputAppend("note", "gone", 0, "bad");
+ runner.appendBuffer("note", "gone", 0, "bad");
+ runner.updateBuffer("note", "present", 0, InterpreterResult.Type.TEXT,
"current");
+ runner.run();
+ runner.appendBuffer("note", "present", 0, "later");
+ runner.run();
+ verify(listener).onOutputUpdated("note", "present", 0,
+ InterpreterResult.Type.TEXT, "current");
+ verify(listener).onOutputAppend("note", "present", 0, "later");
+ }
- String loggerString = "Processing size for buffered append-output is high:
" +
- (data.length() * numEvents) + " characters.";
- assertEquals(loggerString, sizeWarnLogEntry.getMessage());
+ @Test
+ void concurrentDrainCannotOvertakeInFlightCallback() throws Exception {
+ RemoteInterpreterProcessListener listener =
mock(RemoteInterpreterProcessListener.class);
+ AppendOutputRunner runner = new AppendOutputRunner(listener);
+ CountDownLatch entered = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ doAnswer(invocation -> {
+ entered.countDown();
+ assertTrue(release.await(5, TimeUnit.SECONDS));
+ return null;
+ }).when(listener).onOutputAppend("note", "para", 0, "old");
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ runner.appendBuffer("note", "para", 0, "old");
+ Future<?> first = executor.submit(runner);
+ assertTrue(entered.await(5, TimeUnit.SECONDS));
+ runner.updateBuffer("note", "para", 0, InterpreterResult.Type.TEXT,
"new");
+ Future<?> second = executor.submit(runner);
+ assertThrows(TimeoutException.class, () -> second.get(100,
TimeUnit.MILLISECONDS));
+ release.countDown();
+ first.get(5, TimeUnit.SECONDS);
+ second.get(5, TimeUnit.SECONDS);
+ InOrder order = inOrder(listener);
+ order.verify(listener).onOutputAppend("note", "para", 0, "old");
+ order.verify(listener).onOutputUpdated("note", "para", 0,
+ InterpreterResult.Type.TEXT, "new");
+ } finally {
+ release.countDown();
+ executor.shutdownNow();
+ }
}
private class BombardEvents implements Runnable {
@@ -232,4 +322,4 @@ class AppendOutputRunnerTest {
}
}
}
-}
\ No newline at end of file
+}
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerStreamingScopeTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerStreamingScopeTest.java
new file mode 100644
index 0000000000..cdc4d5c7bf
--- /dev/null
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerStreamingScopeTest.java
@@ -0,0 +1,131 @@
+/*
+ * 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.socket;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.List;
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.InterpreterResult;
+import org.apache.zeppelin.interpreter.InterpreterResultMessage;
+import org.apache.zeppelin.notebook.Note;
+import org.apache.zeppelin.notebook.Notebook;
+import org.apache.zeppelin.notebook.Paragraph;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class NotebookServerStreamingScopeTest {
+
+ private NotebookServer server;
+ private Note note;
+ private ConnectionManager connections;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ ZeppelinConfiguration conf = mock(ZeppelinConfiguration.class);
+ when(conf.getBoolean(
+
ZeppelinConfiguration.ConfVars.ZEPPELIN_WEBSOCKET_PARAGRAPH_STATUS_PROGRESS))
+ .thenReturn(true);
+ note = new Note();
+ note.setId("note");
+ note.getParagraphs().add(new Paragraph("para", note, null));
+ Notebook notebook = mock(Notebook.class);
+ doAnswer(invocation -> {
+ Notebook.NoteProcessor<?> processor = invocation.getArgument(1);
+ return processor.process(note);
+ }).when(notebook).processNote(eq("note"), any());
+ connections = mock(ConnectionManager.class);
+ server = new NotebookServer();
+ server.setZeppelinConfiguration(conf);
+ server.setNotebook(() -> notebook);
+ server.setConnectionManager(connections);
+ }
+
+ @Test
+ void sharedNoteReceivesIncrementalOutput() {
+ server.onOutputAppend("note", "para", 0, "append");
+ server.onOutputUpdated("note", "para", 0, InterpreterResult.Type.TEXT,
"update");
+
+ verify(connections, times(2)).broadcast(eq("note"), any());
+ }
+
+ @Test
+ void personalizedNoteDoesNotReceiveUnownedIncrementalOutput() {
+ note.setPersonalizedMode(true);
+
+ server.onOutputAppend("note", "para", 0, "private append");
+ server.onOutputUpdated("note", "para", 0, InterpreterResult.Type.TEXT,
"private update");
+
+ verify(connections, never()).broadcast(eq("note"), any());
+ verify(connections, never()).multicastToUser(any(), any());
+ }
+
+ @Test
+ void personalizedCheckpointDoesNotExposeUnownedOutputToOtherUsers() {
+ note.setPersonalizedMode(true);
+
+ server.onOutputUpdated("note", "para", 0, InterpreterResult.Type.TEXT,
"private update");
+ server.checkpointOutput("note", "para");
+
+ InterpreterResult otherUserResult =
+ note.getParagraph("para").getUserParagraph("other").getReturn();
+ List<InterpreterResultMessage> otherUserMessages =
+ otherUserResult == null ? Collections.emptyList() :
otherUserResult.message();
+ assertTrue(otherUserMessages.stream().noneMatch(m ->
m.getData().contains("private update")),
+ "another user's paragraph sees the checkpointed output: " +
otherUserMessages);
+ }
+
+ @Test
+ void personalizedUnownedClearPreservesSharedOutputForFutureUsers() {
+ note.setPersonalizedMode(true);
+ Paragraph sharedParagraph = note.getParagraph("para");
+ sharedParagraph.setResult(
+ new InterpreterResult(InterpreterResult.Code.SUCCESS, "shared
result"));
+ sharedParagraph.updateOutputBuffer(0, InterpreterResult.Type.TEXT,
"buffered result");
+ sharedParagraph.getUserParagraph("existing");
+
+ server.onOutputClear("note", "para");
+
+ InterpreterResult futureUserResult =
+ sharedParagraph.getUserParagraph("future").getReturn();
+ assertNotNull(futureUserResult,
+ "unowned clear removed the shared output inherited by a future user");
+ assertEquals(1, futureUserResult.message().size());
+ assertEquals("shared result", futureUserResult.message().get(0).getData());
+
+ sharedParagraph.checkpointOutput();
+ InterpreterResult checkpointedFutureUserResult =
+ sharedParagraph.getUserParagraph("checkpointed-future").getReturn();
+ assertNotNull(checkpointedFutureUserResult,
+ "unowned clear removed the shared output buffer used by checkpoint");
+ assertEquals(1, checkpointedFutureUserResult.message().size());
+ assertEquals("buffered result",
checkpointedFutureUserResult.message().get(0).getData());
+ verify(connections, never()).broadcast(eq("note"), any());
+ verify(connections, never()).multicastToUser(any(), any());
+ }
+}
diff --git a/zeppelin-web-angular/e2e/models/published-paragraph-page.ts
b/zeppelin-web-angular/e2e/models/published-paragraph-page.ts
index e63350da02..8916fef0f7 100644
--- a/zeppelin-web-angular/e2e/models/published-paragraph-page.ts
+++ b/zeppelin-web-angular/e2e/models/published-paragraph-page.ts
@@ -16,6 +16,7 @@ import { BasePage } from './base-page';
export class PublishedParagraphPage extends BasePage {
readonly confirmationModal: Locator;
+ readonly textOutput: Locator;
readonly angularRenderer: Locator;
readonly reactWidget: Locator;
readonly reactWidgetOrEmptyState: Locator;
@@ -30,6 +31,7 @@ export class PublishedParagraphPage extends BasePage {
// The result count is 0 in both modes, so dynamic-forms is the
discriminator: it renders only in Angular mode.
this.angularRenderer =
page.locator('zeppelin-notebook-paragraph-dynamic-forms');
this.reactWidget =
page.locator('[data-testid="react-published-paragraph"]');
+ this.textOutput = page.locator('zeppelin-publish-paragraph pre');
// Without paragraph data the remote mounts an <Empty>, so tests that only
assert "React took over" accept either.
this.reactWidgetOrEmptyState =
this.reactWidget.or(page.locator('.ant-alert'));
}
diff --git a/zeppelin-web-angular/e2e/scenarios/notebook-parity.json
b/zeppelin-web-angular/e2e/scenarios/notebook-parity.json
index 9ee808f907..7da8c38817 100644
--- a/zeppelin-web-angular/e2e/scenarios/notebook-parity.json
+++ b/zeppelin-web-angular/e2e/scenarios/notebook-parity.json
@@ -401,6 +401,86 @@
],
"verificationEvidence": []
},
+ {
+ "id": "NB-PARITY-022",
+ "name": "Streaming interpreter output accumulates while a paragraph is
running",
+ "area": "result",
+ "preconditions": [
+ "A disposable notebook has a shell paragraph that prints several
chunks separated by delays.",
+ "The server streams paragraph output
(zeppelin.websocket.paragraph_status_progress.enable is true)."
+ ],
+ "action": "Run the paragraph and observe the result panel before and
after the paragraph finishes.",
+ "observableOutcomes": [
+ {
+ "id": "NB-PARITY-022-OUTCOME-001",
+ "description": "The first output chunk is visible while the
paragraph status is still RUNNING."
+ },
+ {
+ "id": "NB-PARITY-022-OUTCOME-002",
+ "description": "Later chunks are appended after the earlier chunks
instead of replacing them while the paragraph remains RUNNING."
+ },
+ {
+ "id": "NB-PARITY-022-OUTCOME-003",
+ "description": "The FINISHED result contains every chunk exactly
once, in emission order."
+ }
+ ],
+ "interpreter": "sh",
+ "roleExpectations": {
+ "owner": "allow",
+ "writer": "allow",
+ "reader": "deny",
+ "runner": "allow"
+ },
+ "roleVerification": {
+ "owner": "unverified",
+ "writer": "unverified",
+ "reader": "unverified",
+ "runner": "unverified"
+ },
+ "coverage": {
+ "status": "covered",
+ "tests": [
+ {
+ "path":
"zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts",
+ "tag": "@NB-PARITY-022"
+ }
+ ],
+ "issues": [],
+ "uncoveredOutcomes": []
+ },
+ "implementationEvidence": [
+ {
+ "path":
"zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.ts",
+ "symbol": "ParagraphOutputState"
+ },
+ {
+ "path":
"zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts",
+ "symbol": "ParagraphBase.onParagraphAppendOutput"
+ },
+ {
+ "path":
"zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts",
+ "symbol": "ParagraphBase.onParagraphUpdateOutput"
+ },
+ {
+ "path":
"zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java",
+ "symbol": "AppendOutputRunner"
+ }
+ ],
+ "verificationEvidence": [
+ {
+ "path":
"zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.spec.ts",
+ "symbol": "ParagraphOutputState"
+ },
+ {
+ "path":
"zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-stream.capture.json",
+ "symbol": "enabled"
+ },
+ {
+ "path": "zeppelin-web-angular/e2e/models/notebook-paragraph-page.ts",
+ "symbol": "NotebookParagraphPage"
+ }
+ ]
+ },
{
"id": "NB-PARITY-050",
"name": "Notebook editor persists the latest text after typing stops",
diff --git a/zeppelin-web-angular/e2e/scenarios/notebook-parity.md
b/zeppelin-web-angular/e2e/scenarios/notebook-parity.md
index 3da42b6e89..e90d35f5b8 100644
--- a/zeppelin-web-angular/e2e/scenarios/notebook-parity.md
+++ b/zeppelin-web-angular/e2e/scenarios/notebook-parity.md
@@ -32,6 +32,7 @@ Coverage note: `covered` mechanically means this registry
points to a matching e
| NB-PARITY-010 | editor | History inline completion can be dismissed without
losing editor focus | covered | not-applicable |
zeppelin-web-angular/e2e/tests/notebook/inline-completion.spec.ts<br>@NB-PARITY-010
| |
| NB-PARITY-011 | editor | The second Escape after inline completion dismissal
blurs the editor | covered | not-applicable |
zeppelin-web-angular/e2e/tests/notebook/inline-completion.spec.ts<br>@NB-PARITY-011
| |
| NB-PARITY-021 | result | Text and table result displays preserve output
semantics after paragraph execution | partial | owner: allow<br>writer:
allow<br>reader: deny<br>runner: allow |
zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts<br>@NB-PARITY-021
| ZEPPELIN-6514, ZEPPELIN-6516 |
+| NB-PARITY-022 | result | Streaming interpreter output accumulates while a
paragraph is running | covered | owner: allow<br>writer: allow<br>reader:
deny<br>runner: allow |
zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts<br>@NB-PARITY-022
| |
| NB-PARITY-050 | persistence | Notebook editor persists the latest text after
typing stops | gap | owner: allow<br>writer: allow<br>reader: deny<br>runner:
not-applicable | | ZEPPELIN-6661 |
| NB-PARITY-051 | persistence | Notebook editor does not lose an edit made
while a prior save is in flight | gap | owner: allow<br>writer:
allow<br>reader: deny<br>runner: not-applicable | | ZEPPELIN-6661 |
| NB-PARITY-060 | theme | Notebook honors host theme selection | gap |
not-applicable | | ZEPPELIN-6640 |
@@ -135,6 +136,18 @@ Coverage note: `covered` mechanically means this registry
points to a matching e
- Verification evidence: not-applicable
- Uncovered outcomes: NB-PARITY-021-OUTCOME-002: The UI offers every display
mode that the Angular notebook exposes for the returned result type.
NB-PARITY-021-OUTCOME-003: Visualization control changes preserve the field
mapping from result columns to configured dimensions or measures.
NB-PARITY-021-OUTCOME-004: The paragraph's persisted config reflects the
resulting configuration object after a visualization option changes.
NB-PARITY-021-OUTCOME-005: Text and table results expose acces [...]
+### NB-PARITY-022 Streaming interpreter output accumulates while a paragraph
is running
+
+- Area: result
+- Coverage: covered
+- Interpreter: sh
+- Role verification: owner: unverified; writer: unverified; reader:
unverified; runner: unverified
+- Preconditions: A disposable notebook has a shell paragraph that prints
several chunks separated by delays. The server streams paragraph output
(zeppelin.websocket.paragraph_status_progress.enable is true).
+- Action: Run the paragraph and observe the result panel before and after the
paragraph finishes.
+- Observable outcomes: NB-PARITY-022-OUTCOME-001: The first output chunk is
visible while the paragraph status is still RUNNING. NB-PARITY-022-OUTCOME-002:
Later chunks are appended after the earlier chunks instead of replacing them
while the paragraph remains RUNNING. NB-PARITY-022-OUTCOME-003: The FINISHED
result contains every chunk exactly once, in emission order.
+- Implementation evidence:
zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.ts
(ParagraphOutputState);
zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts
(ParagraphBase.onParagraphAppendOutput);
zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts
(ParagraphBase.onParagraphUpdateOutput);
zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java
(AppendOutputRunner)
+- Verification evidence:
zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.spec.ts
(ParagraphOutputState);
zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-stream.capture.json
(enabled); zeppelin-web-angular/e2e/models/notebook-paragraph-page.ts
(NotebookParagraphPage)
+
### NB-PARITY-050 Notebook editor persists the latest text after typing stops
- Area: persistence
diff --git
a/zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts
b/zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts
index c238b2269b..6d01796326 100644
---
a/zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts
+++
b/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,37 @@ test.describe('Notebook Paragraph Functionality', () => {
await expect(paragraphPage.resultDisplay).not.toBeEmpty();
});
+ test(
+ 'should accumulate interpreter output while the paragraph is running',
+ { tag: '@NB-PARITY-022' },
+ 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');
+ await
expect(paragraphPage.resultDisplay).toHaveText('first\nsecond\nthird\n');
+ });
+ }
+ );
+
test('should display dynamic forms', async ({ page }) => {
test.skip(!!process.env.CI, 'Dynamic form tests require a Spark
interpreter — skipped on CI');
diff --git
a/zeppelin-web-angular/e2e/tests/notebook/published/published-streaming.spec.ts
b/zeppelin-web-angular/e2e/tests/notebook/published/published-streaming.spec.ts
new file mode 100644
index 0000000000..208a872142
--- /dev/null
+++
b/zeppelin-web-angular/e2e/tests/notebook/published/published-streaming.spec.ts
@@ -0,0 +1,120 @@
+/*
+ * 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 { createServer } from 'node:http';
+
+import { expect, test } from '@playwright/test';
+import { NotebookParagraphPage } from
'../../../models/notebook-paragraph-page';
+import { PublishedParagraphPage } from
'../../../models/published-paragraph-page';
+import {
+ addPageAnnotationBeforeEach,
+ createTestNotebook,
+ PAGES,
+ setParagraphText,
+ waitForZeppelinReady
+} from '../../../utils';
+
+test.describe('Published paragraph streaming', () => {
+ // JUSTIFIED: shared owner and notebook state must remain within one worker.
+ test.describe.configure({ mode: 'default' });
+ addPageAnnotationBeforeEach(PAGES.WORKSPACE.PUBLISHED_PARAGRAPH);
+
+ let owner: NotebookParagraphPage;
+ let notebook: { noteId: string; paragraphId: string };
+
+ test.beforeEach(async ({ page }) => {
+ owner = new NotebookParagraphPage(page);
+ await page.goto('/#/');
+ await waitForZeppelinReady(page);
+ notebook = await createTestNotebook(page);
+ });
+
+ for (const react of [false, true]) {
+ test(`accumulates live output and preserves the exact terminal snapshot
(${react ? 'React' : 'Angular'})`, async ({
+ page,
+ context
+ }) => {
+ const viewer = await context.newPage();
+ const published = new PublishedParagraphPage(viewer);
+
+ // The local shell waits for the viewer before emitting the next chunk.
+ const release: Array<() => void> = [];
+ const gates = [0, 1].map(index => new Promise<void>(resolve =>
(release[index] = resolve)));
+ const outputGate = createServer((request, response) => {
+ const gate = gates[Number(request.url?.slice(1))];
+ if (!gate) {
+ response.writeHead(404).end();
+ return;
+ }
+ void gate.then(() => response.end());
+ });
+ await new Promise<void>((resolve, reject) => {
+ outputGate.once('error', reject);
+ outputGate.listen(0, '127.0.0.1', resolve);
+ });
+
+ try {
+ const address = outputGate.address();
+ if (!address || typeof address === 'string') {
+ throw new Error('Failed to bind output gate');
+ }
+ const waitForViewer = `curl --noproxy '*' --fail --silent --show-error
--max-time 180 http://127.0.0.1:${address.port}`;
+
+ await test.step('Given a published viewer of a shell paragraph with
controlled output', async () => {
+ await setParagraphText(
+ page,
+ notebook.noteId,
+ notebook.paragraphId,
+ `%sh\nset -e\necho first\n${waitForViewer}/0\necho
second\n${waitForViewer}/1\necho third`
+ );
+ await page.goto(`/#/notebook/${notebook.noteId}`);
+ await expect(owner.paragraphContainer).toBeVisible();
+ await
viewer.goto(`/#/notebook/${notebook.noteId}/paragraph/${notebook.paragraphId}?react=${react}`);
+ await waitForZeppelinReady(viewer);
+ await expect(published.confirmationModal).toBeVisible();
+ await published.cancelButton.click();
+ });
+
+ await test.step('When the owner runs the paragraph, published output
accumulates while RUNNING', async () => {
+ await owner.runParagraph();
+ await expect(published.textOutput).toHaveText(/^first\n$/);
+ await expect(published.textOutput).toBeVisible();
+ await expect(owner.status).toHaveText('RUNNING');
+ release[0]();
+ await expect(published.textOutput).toHaveText(/^first\nsecond\n$/);
+ await expect(published.textOutput).toBeVisible();
+ await expect(owner.status).toHaveText('RUNNING');
+ release[1]();
+ });
+
+ await test.step('Then the terminal snapshot contains every chunk
exactly once', async () => {
+ await expect(owner.status).toHaveText('FINISHED');
+ await
expect(published.textOutput).toHaveText(/^first\nsecond\nthird\n$/);
+ await expect(published.reactWidget).toHaveCount(react ? 1 : 0);
+ });
+
+ await test.step('And a new run replaces the previous stream', async ()
=> {
+ await setParagraphText(page, notebook.noteId, notebook.paragraphId,
'%sh\necho rerun');
+ await owner.runParagraph();
+ await expect(owner.status).toHaveText('FINISHED');
+ await expect(published.textOutput).toHaveText(/^rerun\n$/);
+ });
+ } finally {
+ await new Promise<void>((resolve, reject) => {
+ outputGate.close(error => (error ? reject(error) : resolve()));
+ outputGate.closeAllConnections();
+ });
+ await viewer.close();
+ }
+ });
+ }
+});
diff --git
a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.spec.ts
b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.spec.ts
new file mode 100644
index 0000000000..ec7797ecba
--- /dev/null
+++
b/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>();
+});
diff --git
a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.ts
b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.ts
index f86dbdb2b1..6c023a0923 100644
---
a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.ts
+++
b/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;
diff --git
a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-paragraph.interface.ts
b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-paragraph.interface.ts
index f75cd1f5f3..f9e01351e2 100644
---
a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-paragraph.interface.ts
+++
b/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;
diff --git
a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts
b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts
new file mode 100644
index 0000000000..34d5cc4418
--- /dev/null
+++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts
@@ -0,0 +1,247 @@
+/*
+ * 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 { ChangeDetectorRef } from '@angular/core';
+import { DatasetType, Message, ParagraphItem } from '@zeppelin/sdk';
+import { EMPTY } from 'rxjs';
+import { describe, expect, it, vi } from 'vitest';
+
+import { AngularContextManager } from './angular-context-manager';
+import { ParagraphBase } from './paragraph-base';
+
+class TestParagraph extends ParagraphBase {
+ changeColWidth = vi.fn();
+ updateParagraphResult = vi.fn();
+ protected currentNoteId = 'note';
+
+ hydrate(snapshot: ParagraphItem) {
+ this.setParagraphSnapshot(snapshot);
+ }
+
+ constructor(paragraph: ParagraphItem) {
+ super(
+ { receive: () => EMPTY } as unknown as Message,
+ { isParagraphRunning: item => item.status === 'RUNNING',
isEntireNoteRunning: () => false },
+ {} as AngularContextManager,
+ { markForCheck: vi.fn() } as unknown as ChangeDetectorRef
+ );
+ this.paragraph = paragraph;
+ }
+}
+
+const paragraph = (id: string, status = 'RUNNING', dateStarted =
'2026-01-01T00:00:00Z'): ParagraphItem => ({
+ id,
+ status,
+ dateStarted,
+ text: '',
+ user: 'anonymous',
+ dateUpdated: '',
+ dateCreated: '',
+ config: {},
+ settings: { params: {}, forms: {} },
+ apps: [],
+ progressUpdateIntervalMs: 500,
+ jobName: '',
+ aborted: false,
+ lineNumbers: false,
+ fontSize: 9
+});
+
+const beginOutput = (component: TestParagraph) =>
+ component.onParagraphUpdateOutput({
+ noteId: 'note',
+ paragraphId: 'A',
+ index: 0,
+ type: DatasetType.TEXT,
+ data: 'first\n'
+ });
+
+const appendOutput = (component: TestParagraph) =>
+ component.onParagraphAppendOutput({ noteId: 'note', paragraphId: 'A', index:
0, data: 'second\n' });
+
+describe('ParagraphBase streaming state isolation', () => {
+ it.each(['append', 'update'])('ignores %s from another note with the same
paragraph ID', kind => {
+ const component = new TestParagraph(paragraph('A'));
+ beginOutput(component);
+ if (kind === 'append') {
+ component.onParagraphAppendOutput({ noteId: 'other-note', paragraphId:
'A', index: 0, data: 'foreign' });
+ } else {
+ component.onParagraphUpdateOutput({
+ noteId: 'other-note',
+ paragraphId: 'A',
+ index: 0,
+ type: DatasetType.HTML,
+ data: 'foreign'
+ });
+ }
+ expect(component.results).toEqual([{ type: DatasetType.TEXT, data:
'first\n' }]);
+ component.ngOnDestroy();
+ });
+
+ it.each(['FINISHED', 'ERROR', 'ABORT', 'RUNNING', 'PENDING'])(
+ 'keeps accumulating output when another paragraph becomes %s',
+ status => {
+ const component = new TestParagraph(paragraph('A'));
+ beginOutput(component);
+
+ component.paragraphData({ paragraph: paragraph('B', status,
'2026-01-01T00:00:01Z') });
+ appendOutput(component);
+
+ expect(component.results).toEqual([{ type: DatasetType.TEXT, data:
'first\nsecond\n' }]);
+ expect(component.paragraph?.id).toBe('A');
+ component.ngOnDestroy();
+ }
+ );
+
+ it('ignores late output after its own terminal snapshot', () => {
+ const component = new TestParagraph(paragraph('A'));
+ beginOutput(component);
+ const finished = { ...paragraph('A', 'FINISHED'), results: { msg: [{ type:
DatasetType.TEXT, data: 'final\n' }] } };
+
+ component.paragraphData({ paragraph: finished });
+ appendOutput(component);
+
+ expect(component.results).toEqual([{ type: DatasetType.TEXT, data:
'final\n' }]);
+ component.ngOnDestroy();
+ });
+
+ it('accepts fresh output when its own paragraph starts a new run', () => {
+ const component = new TestParagraph(paragraph('A'));
+ beginOutput(component);
+ component.paragraphData({ paragraph: paragraph('A', 'FINISHED') });
+ component.paragraphData({ paragraph: paragraph('A', 'RUNNING',
'2026-01-01T00:00:01Z') });
+
+ beginOutput(component);
+ appendOutput(component);
+
+ expect(component.results).toEqual([{ type: DatasetType.TEXT, data:
'first\nsecond\n' }]);
+ component.ngOnDestroy();
+ });
+});
+
+describe('ParagraphBase streaming boundaries', () => {
+ it('preserves a saved revision during live output and lifecycle messages',
() => {
+ const component = new TestParagraph(paragraph('A', 'FINISHED'));
+ const saved = [{ type: DatasetType.TEXT, data: 'saved revision' }];
+ component.paragraph!.results = { msg: saved };
+ component.results = saved;
+ component.revisionView = true;
+
+ component.paragraphData({ paragraph: paragraph('A', 'RUNNING',
'2026-01-01T00:00:01Z') });
+ beginOutput(component);
+ appendOutput(component);
+ component.paragraphData({ paragraph: paragraph('A') });
+
+ expect(component.results).toEqual(saved);
+ expect(component.paragraph?.status).toBe('FINISHED');
+ component.revisionView = false;
+ beginOutput(component);
+ expect(component.results).toEqual(saved);
+ component.ngOnDestroy();
+ });
+
+ it('does not refresh an unchanged paragraph for an omitted run timestamp',
() => {
+ const component = new TestParagraph(paragraph('A'));
+ const update = vi.spyOn(component, 'updateParagraph');
+ const statusOnly = paragraph('A');
+ delete statusOnly.dateStarted;
+
+ component.paragraphData({ paragraph: statusOnly });
+ component.paragraphData({ paragraph: statusOnly });
+
+ expect(update).not.toHaveBeenCalled();
+ expect(component.paragraph?.dateStarted).toBe('2026-01-01T00:00:00Z');
+ component.paragraphData({ paragraph: paragraph('A', 'RUNNING',
'2026-01-01T00:00:01Z') });
+ expect(update).toHaveBeenCalledOnce();
+ expect(component.paragraph?.dateStarted).toBe('2026-01-01T00:00:01Z');
+ component.ngOnDestroy();
+ });
+
+ it('publishes the final result when terminal status arrives before its
snapshot', () => {
+ const component = new TestParagraph(paragraph('A'));
+ beginOutput(component);
+ component.onParagraphStatus({ id: 'A', status: 'FINISHED' });
+ const finished = paragraph('A', 'FINISHED');
+ delete finished.dateStarted;
+ finished.results = { msg: [{ type: DatasetType.TEXT, data:
'first\nfinal\n' }] };
+
+ component.paragraphData({ paragraph: finished });
+
+ expect(component.results).toEqual([{ type: DatasetType.TEXT, data:
'first\nfinal\n' }]);
+ expect(component.updateParagraphResult).toHaveBeenCalledWith(0,
expect.anything(), {
+ type: DatasetType.TEXT,
+ data: 'first\nfinal\n'
+ });
+ component.ngOnDestroy();
+ });
+
+ it('preserves streamed types across a status snapshot without a run
timestamp', () => {
+ const component = new TestParagraph(paragraph('A', 'PENDING'));
+ beginOutput(component);
+ const running = paragraph('A');
+ delete running.dateStarted;
+ component.paragraphData({ paragraph: running });
+ component.paragraphData({ paragraph: paragraph('A') });
+ appendOutput(component);
+
+ expect(component.results).toEqual([{ type: DatasetType.TEXT, data:
'first\nsecond\n' }]);
+ expect(component.paragraph?.dateStarted).toBe('2026-01-01T00:00:00Z');
+ component.ngOnDestroy();
+ });
+
+ it('preserves streaming output across ordinary RUNNING and unrelated
paragraph snapshots', () => {
+ const component = new TestParagraph(paragraph('A'));
+ beginOutput(component);
+ component.paragraphData({ paragraph: paragraph('A') });
+ component.paragraphData({ paragraph: paragraph('B') });
+ appendOutput(component);
+ expect(component.results).toEqual([{ type: DatasetType.TEXT, data:
'first\nsecond\n' }]);
+ component.ngOnDestroy();
+ });
+
+ it('buffers sparse result slots without rendering holes or changing server
indexes', () => {
+ const component = new TestParagraph(paragraph('A'));
+ component.onParagraphAppendOutput({ noteId: 'note', paragraphId: 'A',
index: 2, data: ' tail' });
+ component.setResults(component.paragraph!);
+ component.onParagraphUpdateOutput({
+ noteId: 'note',
+ paragraphId: 'A',
+ index: 2,
+ type: DatasetType.TEXT,
+ data: 'third'
+ });
+ expect(component.results).toEqual([]);
+ expect(component.updateParagraphResult).not.toHaveBeenCalled();
+ component.setResults(component.paragraph!);
+ beginOutput(component);
+ expect(component.results).toEqual([{ type: DatasetType.TEXT, data:
'first\n' }]);
+ component.onParagraphUpdateOutput({
+ noteId: 'note',
+ paragraphId: 'A',
+ index: 1,
+ type: DatasetType.TABLE,
+ data: 'second'
+ });
+ expect(component.results).toEqual([
+ { type: DatasetType.TEXT, data: 'first\n' },
+ { type: DatasetType.TABLE, data: 'second' },
+ { type: DatasetType.TEXT, data: 'third' }
+ ]);
+ expect(component.updateParagraphResult).toHaveBeenCalledWith(2,
expect.anything(), {
+ type: DatasetType.TEXT,
+ data: 'third'
+ });
+ component.onParagraphAppendOutput({ noteId: 'note', paragraphId: 'A',
index: 2, data: ' end' });
+ expect(component.results[2].data).toBe('third end');
+ component.ngOnDestroy();
+ });
+});
diff --git a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts
b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts
index 4e7c0e0fde..b9e514abd6 100644
--- a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts
+++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts
@@ -26,12 +26,13 @@ import {
ParagraphIResultsMsgItem
} from '@zeppelin/sdk';
-import * as DiffMatchPatch from 'diff-match-patch';
+import { diff_match_patch as DiffMatchPatch } from 'diff-match-patch';
import { isEmpty, isEqual } from 'lodash';
import { MessageListener, MessageListenersManager } from
'../message-listener/message-listener';
import { AngularContextManager } from './angular-context-manager';
import { NoteStatus } from './note-status';
+import { ParagraphOutputState } from './paragraph-output-state';
export const ParagraphStatus = {
READY: 'READY',
@@ -42,6 +43,9 @@ export const ParagraphStatus = {
ERROR: 'ERROR'
};
+const isTerminalParagraphStatus = (status?: string): boolean =>
+ status === ParagraphStatus.FINISHED || status === ParagraphStatus.ABORT ||
status === ParagraphStatus.ERROR;
+
export abstract class ParagraphBase extends MessageListenersManager {
paragraph?: ParagraphItem;
dirtyText?: string;
@@ -58,6 +62,7 @@ export abstract class ParagraphBase extends
MessageListenersManager {
params: {},
forms: {}
};
+ private readonly outputState = new ParagraphOutputState();
constructor(
public messageService: Message,
@@ -68,6 +73,8 @@ export abstract class ParagraphBase extends
MessageListenersManager {
super(messageService);
}
+ protected abstract get currentNoteId(): string | null | undefined;
+
abstract changeColWidth(needCommit: boolean, updateResult?: boolean): void;
@MessageListener(OP.PROGRESS)
@@ -114,16 +121,54 @@ export abstract class ParagraphBase extends
MessageListenersManager {
}
}
+ @MessageListener(OP.PARAGRAPH_APPEND_OUTPUT)
+ onParagraphAppendOutput(data:
MessageReceiveDataTypeMap[OP.PARAGRAPH_APPEND_OUTPUT]) {
+ if (this.revisionView || data.noteId !== this.currentNoteId ||
data.paragraphId !== this.paragraph?.id) {
+ return;
+ }
+ this.initializeOutputState();
+ const result = this.outputState.append(data.index, data.data);
+ if (result) {
+ this.applyStreamingResult(data.index);
+ }
+ }
+
+ @MessageListener(OP.PARAGRAPH_UPDATE_OUTPUT)
+ onParagraphUpdateOutput(data:
MessageReceiveDataTypeMap[OP.PARAGRAPH_UPDATE_OUTPUT]) {
+ if (this.revisionView || data.noteId !== this.currentNoteId ||
data.paragraphId !== this.paragraph?.id) {
+ return;
+ }
+ this.initializeOutputState();
+ const result = this.outputState.update(data.index, data.type, data.data);
+ if (result) {
+ this.applyStreamingResult(data.index);
+ }
+ }
+
@MessageListener(OP.PARAGRAPH)
paragraphData(data: MessageReceiveDataTypeMap[OP.PARAGRAPH]) {
const oldPara = this.paragraph;
if (!oldPara) {
- throw new Error('paragraph is not defined');
+ return;
}
const newPara = data.paragraph;
+ if (this.revisionView || newPara.id !== oldPara.id) {
+ return;
+ }
if (!newPara.results) {
newPara.results = {};
}
+ const oldRunActive = oldPara.status === ParagraphStatus.PENDING ||
oldPara.status === ParagraphStatus.RUNNING;
+ const newRunActive = newPara.status === ParagraphStatus.PENDING ||
newPara.status === ParagraphStatus.RUNNING;
+ const runChanged =
+ newPara.dateStarted != null && oldPara.dateStarted != null &&
newPara.dateStarted !== oldPara.dateStarted;
+ if (newRunActive && (!oldRunActive || runChanged)) {
+ this.outputState.reset();
+ }
+ // Close the stream before publishing the terminal snapshot.
+ if (isTerminalParagraphStatus(newPara.status)) {
+ this.outputState.finish(newPara.results?.msg);
+ }
if (this.isUpdateRequired(oldPara, newPara)) {
this.updateParagraph(oldPara, newPara, () => {
if (newPara.results && newPara.results.msg) {
@@ -186,6 +231,32 @@ export abstract class ParagraphBase extends
MessageListenersManager {
}
}
+ private initializeOutputState(): void {
+ if (!this.outputState.isInitialized) {
+ this.outputState.reset(this.results,
isTerminalParagraphStatus(this.paragraph?.status));
+ }
+ }
+
+ private applyStreamingResult(index: number): void {
+ if (!this.paragraph) {
+ return;
+ }
+ const previousLength = this.results.length;
+ const results = this.outputState.snapshot();
+ if (!this.paragraph.results) {
+ this.paragraph.results = {};
+ }
+ this.paragraph.results.msg = results;
+ this.results = results;
+ results.forEach((visibleResult, visibleIndex) => {
+ if (visibleIndex === index || visibleIndex >= previousLength) {
+ const config = this.paragraph!.config.results?.[visibleIndex] ?? {
graph: new GraphConfig() };
+ this.updateParagraphResult(visibleIndex, config, visibleResult);
+ }
+ });
+ this.cdr.markForCheck();
+ }
+
updateParagraph(oldPara: ParagraphItem, newPara: ParagraphItem,
updateCallback: () => void) {
// 1. can't update on revision view
if (!this.revisionView) {
@@ -221,12 +292,13 @@ export abstract class ParagraphBase extends
MessageListenersManager {
(newPara.dateCreated !== oldPara.dateCreated ||
newPara.text !== oldPara.text ||
newPara.dateFinished !== oldPara.dateFinished ||
- newPara.dateStarted !== oldPara.dateStarted ||
+ (newPara.dateStarted != null && newPara.dateStarted !==
oldPara.dateStarted) ||
newPara.dateUpdated !== oldPara.dateUpdated ||
newPara.status !== oldPara.status ||
newPara.jobName !== oldPara.jobName ||
newPara.title !== oldPara.title ||
isEmpty(newPara.results) !== isEmpty(oldPara.results) ||
+ (isTerminalParagraphStatus(newPara.status) &&
!isEqual(newPara.results?.msg, oldPara.results?.msg)) ||
newPara.errorMessage !== oldPara.errorMessage ||
!isEqual(newPara.settings, oldPara.settings) ||
!isEqual(newPara.config, oldPara.config) ||
@@ -270,7 +342,10 @@ export abstract class ParagraphBase extends
MessageListenersManager {
this.paragraph.dateUpdated = newPara.dateUpdated;
this.paragraph.dateCreated = newPara.dateCreated;
this.paragraph.dateFinished = newPara.dateFinished;
- this.paragraph.dateStarted = newPara.dateStarted;
+ // Status-only snapshots can omit the start time of the current run.
+ if (newPara.dateStarted != null) {
+ this.paragraph.dateStarted = newPara.dateStarted;
+ }
this.paragraph.errorMessage = newPara.errorMessage;
this.paragraph.jobName = newPara.jobName;
this.paragraph.title = newPara.title;
@@ -361,4 +436,15 @@ export abstract class ParagraphBase extends
MessageListenersManager {
}
this.messageService.cancelParagraph(this.paragraph.id);
}
+ protected setParagraphSnapshot(paragraph: ParagraphItem | undefined): void {
+ this.paragraph = paragraph;
+ this.results = [];
+ this.configs = {};
+ if (paragraph) {
+ this.setResults(paragraph);
+ }
+ const terminal = isTerminalParagraphStatus(paragraph?.status);
+ this.outputState.reset(this.results, terminal);
+ this.cdr.markForCheck();
+ }
}
diff --git
a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.spec.ts
b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.spec.ts
new file mode 100644
index 0000000000..e9ce623d4d
--- /dev/null
+++
b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.spec.ts
@@ -0,0 +1,151 @@
+/*
+ * 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 { DatasetType, ParagraphIResultsMsgItem } from '@zeppelin/sdk';
+import { describe, expect, it } from 'vitest';
+
+import capture from './paragraph-output-stream.capture.json';
+import { ParagraphOutputState } from './paragraph-output-state';
+
+interface CapturedEvent {
+ op: string;
+ data: {
+ index?: number;
+ type?: string;
+ data?: string;
+ paragraph?: {
+ status: string;
+ results?: {
+ msg?: Array<{ type: string; data: string }>;
+ };
+ };
+ };
+}
+
+const capturedType = (type: string): DatasetType => {
+ expect(Object.values(DatasetType)).toContain(type);
+ return type as DatasetType;
+};
+
+const capturedResults = (results: Array<{ type: string; data: string }>):
ParagraphIResultsMsgItem[] =>
+ results.map(result => ({ type: capturedType(result.type), data: result.data
}));
+
+const replay = (state: ParagraphOutputState, events: CapturedEvent[]):
string[] => {
+ const rendered: string[] = [];
+ for (const event of events) {
+ if (event.op === 'PARAGRAPH_UPDATE_OUTPUT') {
+ const result = state.update(event.data.index!,
capturedType(event.data.type!), event.data.data!);
+ if (result) {
+ rendered.push(result.data);
+ }
+ } else if (event.op === 'PARAGRAPH_APPEND_OUTPUT') {
+ const result = state.append(event.data.index!, event.data.data!);
+ if (result) {
+ rendered.push(result.data);
+ }
+ } else if (event.op === 'PARAGRAPH') {
+ state.finish(capturedResults(event.data.paragraph?.results?.msg ?? []));
+ }
+ }
+ return rendered;
+};
+
+describe('ParagraphOutputState', () => {
+ it('replays the captured callback and WebSocket order without dropping
output', () => {
+ const state = new ParagraphOutputState();
+ state.reset();
+ const events = capture.enabled.events as CapturedEvent[];
+
+ expect(capture.schemaVersion).toBe(1);
+ expect(events.slice(0, -1).map(({ op, data }) => ({ op, data
}))).toEqual(capture.callbackOrder.events);
+ expect(replay(state, events)).toEqual(['', 'first\n', 'first\nsecond\n',
'first\nsecond\nthird\n']);
+ expect(state.snapshot()).toEqual([{ type: DatasetType.TEXT, data:
'first\nsecond\nthird\n' }]);
+ });
+
+ it('accumulates coalesced APPEND chunks at their result index', () => {
+ const state = new ParagraphOutputState();
+ const appends = capture.enabled.events.filter(event => event.op ===
'PARAGRAPH_APPEND_OUTPUT');
+ state.reset([{ type: DatasetType.TEXT, data: '' }]);
+
+ state.append(0, appends[0].data.data! + appends[1].data.data!);
+ const result = state.append(0, appends[2].data.data!);
+
+ expect(result).toEqual({ type: DatasetType.TEXT, data:
'first\nsecond\nthird\n' });
+ expect(state.snapshot()).toEqual([result]);
+ });
+
+ it('holds APPEND chunks until a typed UPDATE can render them', () => {
+ const state = new ParagraphOutputState();
+ const update = capture.enabled.events.find(event => event.op ===
'PARAGRAPH_UPDATE_OUTPUT')!;
+ const appends = capture.enabled.events.filter(event => event.op ===
'PARAGRAPH_APPEND_OUTPUT');
+ state.reset();
+
+ expect(state.append(0, appends[0].data.data!)).toBeUndefined();
+ expect(state.append(0, appends[1].data.data!)).toBeUndefined();
+
+ expect(state.update(0, capturedType(update.data.type!),
update.data.data!)).toEqual({
+ type: DatasetType.TEXT,
+ data: 'first\nsecond\n'
+ });
+ });
+
+ it('discards pending APPEND chunks when a non-empty UPDATE replaces the
result', () => {
+ const state = new ParagraphOutputState();
+ state.reset();
+
+ expect(state.append(0, 'stale\n')).toBeUndefined();
+ expect(state.update(0, DatasetType.TEXT, 'replacement\n')).toEqual({
+ type: DatasetType.TEXT,
+ data: 'replacement\n'
+ });
+ expect(state.snapshot()).toEqual([{ type: DatasetType.TEXT, data:
'replacement\n' }]);
+ expect(state.update(0, DatasetType.TEXT, '')).toEqual({
+ type: DatasetType.TEXT,
+ data: ''
+ });
+ expect(state.snapshot()).toEqual([{ type: DatasetType.TEXT, data: '' }]);
+ });
+
+ it('uses the terminal snapshot after UPDATE overtakes a queued APPEND', ()
=> {
+ const state = new ParagraphOutputState();
+ state.reset([{ type: DatasetType.TEXT, data: 'stale\n' }]);
+
+ state.update(0, DatasetType.TEXT, 'replacement\n');
+ state.append(0, 'queued-before-update\n');
+ state.finish([{ type: DatasetType.TEXT, data: 'replacement\n' }]);
+
+ expect(state.snapshot()).toEqual([{ type: DatasetType.TEXT, data:
'replacement\n' }]);
+ });
+
+ it('ignores an APPEND observed after the terminal PARAGRAPH', () => {
+ const state = new ParagraphOutputState();
+ const events = capture.enabled.events as CapturedEvent[];
+ const terminal = events.find(event => event.op === 'PARAGRAPH')!;
+ const finalAppend = [...events].reverse().find(event => event.op ===
'PARAGRAPH_APPEND_OUTPUT')!;
+ state.reset([{ type: DatasetType.TEXT, data: 'first\nsecond\n' }]);
+
+ replay(state, [terminal, finalAppend]);
+
+ expect(state.snapshot()).toEqual([{ type: DatasetType.TEXT, data:
'first\nsecond\nthird\n' }]);
+ });
+
+ it('falls back to the terminal snapshot when streaming messages are
disabled', () => {
+ const state = new ParagraphOutputState();
+ const events = capture.disabled.events as CapturedEvent[];
+ state.reset();
+
+
expect(capture.disabled.configuration['zeppelin.websocket.paragraph_status_progress.enable']).toBe(false);
+ expect(events.map(event => event.op)).toEqual(['PARAGRAPH']);
+ expect(() => replay(state, events)).not.toThrow();
+ expect(state.snapshot()).toEqual([{ type: DatasetType.TEXT, data:
'first\nsecond\nthird\n' }]);
+ });
+});
diff --git
a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.ts
b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.ts
new file mode 100644
index 0000000000..93934ededc
--- /dev/null
+++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.ts
@@ -0,0 +1,79 @@
+/*
+ * 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 { DatasetType, ParagraphIResultsMsgItem } from '@zeppelin/sdk';
+
+export class ParagraphOutputState {
+ private results: ParagraphIResultsMsgItem[] = [];
+ private readonly pendingAppends = new Map<number, string>();
+ private initialized = false;
+ private terminal = false;
+
+ get isInitialized(): boolean {
+ return this.initialized;
+ }
+
+ reset(results: ParagraphIResultsMsgItem[] = [], terminal = false): void {
+ this.results = results.map(result => ({ ...result }));
+ this.pendingAppends.clear();
+ this.initialized = true;
+ this.terminal = terminal;
+ }
+
+ finish(results: ParagraphIResultsMsgItem[] = []): void {
+ this.reset(results, true);
+ }
+
+ update(index: number, type: DatasetType, data: string):
ParagraphIResultsMsgItem | undefined {
+ if (this.terminal) {
+ return undefined;
+ }
+
+ // Non-empty UPDATE data replaces stale appends; an empty type declaration
adopts them.
+ const result = {
+ type,
+ data: data === '' ? (this.pendingAppends.get(index) ?? '') : data
+ };
+ this.pendingAppends.delete(index);
+ this.results[index] = result;
+ return result;
+ }
+
+ append(index: number, data: string): ParagraphIResultsMsgItem | undefined {
+ if (this.terminal) {
+ return undefined;
+ }
+
+ const current = this.results[index];
+ if (!current) {
+ this.pendingAppends.set(index, (this.pendingAppends.get(index) ?? '') +
data);
+ return undefined;
+ }
+
+ const result = {
+ ...current,
+ data: current.data + data
+ };
+ this.results[index] = result;
+ return result;
+ }
+
+ snapshot(): ParagraphIResultsMsgItem[] {
+ // Preserve server indexes by waiting for missing slots to acquire their
types.
+ // Keep later slots buffered until those gaps are filled.
+ let length = 0;
+ while (this.results[length]) {
+ length++;
+ }
+ return this.results.slice(0, length);
+ }
+}
diff --git
a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-stream.capture.json
b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-stream.capture.json
new file mode 100644
index 0000000000..655ca5fbba
--- /dev/null
+++
b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-stream.capture.json
@@ -0,0 +1,148 @@
+{
+ "_license": "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.",
+ "schemaVersion": 1,
+ "capture": {
+ "issue": "ZEPPELIN-6659",
+ "zeppelinVersion": "0.13.0-SNAPSHOT",
+ "capturedAt": "2026-09-05T07:15:34Z",
+ "browser": "Chromium",
+ "interpreter": "sh",
+ "paragraph": "echo first; sleep 3; echo second; sleep 5; echo third",
+ "normalization": "noteId, paragraphId, principal, ticket, message ids, and
absolute timestamps are normalized or omitted"
+ },
+ "callbackOrder": {
+ "evidence": "InterpreterResultMessageOutput.java:108-119 emits the initial
typed update before flushing the first append callback",
+ "events": [
+ {
+ "op": "PARAGRAPH_UPDATE_OUTPUT",
+ "data": {
+ "noteId": "<note-id>",
+ "paragraphId": "<paragraph-id>",
+ "index": 0,
+ "type": "TEXT",
+ "data": ""
+ }
+ },
+ {
+ "op": "PARAGRAPH_APPEND_OUTPUT",
+ "data": {
+ "noteId": "<note-id>",
+ "paragraphId": "<paragraph-id>",
+ "index": 0,
+ "data": "first\n"
+ }
+ },
+ {
+ "op": "PARAGRAPH_APPEND_OUTPUT",
+ "data": {
+ "noteId": "<note-id>",
+ "paragraphId": "<paragraph-id>",
+ "index": 0,
+ "data": "second\n"
+ }
+ },
+ {
+ "op": "PARAGRAPH_APPEND_OUTPUT",
+ "data": {
+ "noteId": "<note-id>",
+ "paragraphId": "<paragraph-id>",
+ "index": 0,
+ "data": "third\n"
+ }
+ }
+ ]
+ },
+ "enabled": {
+ "configuration": {
+ "zeppelin.websocket.paragraph_status_progress.enable": true
+ },
+ "events": [
+ {
+ "elapsedMs": 0,
+ "op": "PARAGRAPH_UPDATE_OUTPUT",
+ "data": {
+ "noteId": "<note-id>",
+ "paragraphId": "<paragraph-id>",
+ "index": 0,
+ "type": "TEXT",
+ "data": ""
+ }
+ },
+ {
+ "elapsedMs": 107,
+ "op": "PARAGRAPH_APPEND_OUTPUT",
+ "data": {
+ "noteId": "<note-id>",
+ "paragraphId": "<paragraph-id>",
+ "index": 0,
+ "data": "first\n"
+ }
+ },
+ {
+ "elapsedMs": 2984,
+ "op": "PARAGRAPH_APPEND_OUTPUT",
+ "data": {
+ "noteId": "<note-id>",
+ "paragraphId": "<paragraph-id>",
+ "index": 0,
+ "data": "second\n"
+ }
+ },
+ {
+ "elapsedMs": 7990,
+ "op": "PARAGRAPH_APPEND_OUTPUT",
+ "data": {
+ "noteId": "<note-id>",
+ "paragraphId": "<paragraph-id>",
+ "index": 0,
+ "data": "third\n"
+ }
+ },
+ {
+ "elapsedMs": 8027,
+ "op": "PARAGRAPH",
+ "data": {
+ "paragraph": {
+ "id": "<paragraph-id>",
+ "status": "FINISHED",
+ "results": {
+ "code": "SUCCESS",
+ "msg": [
+ {
+ "type": "TEXT",
+ "data": "first\nsecond\nthird\n"
+ }
+ ]
+ }
+ }
+ }
+ }
+ ]
+ },
+ "disabled": {
+ "configuration": {
+ "zeppelin.websocket.paragraph_status_progress.enable": false
+ },
+ "events": [
+ {
+ "elapsedMs": 8006,
+ "op": "PARAGRAPH",
+ "data": {
+ "paragraph": {
+ "id": "<paragraph-id>",
+ "status": "FINISHED",
+ "results": {
+ "code": "SUCCESS",
+ "msg": [
+ {
+ "type": "TEXT",
+ "data": "first\nsecond\nthird\n"
+ }
+ ]
+ }
+ }
+ }
+ }
+ ]
+ }
+}
diff --git
a/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts
b/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts
index 17c9a4d578..c7365dd3c9 100644
---
a/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts
+++
b/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts
@@ -102,6 +102,10 @@ export class NotebookParagraphComponent
@Input() useReactFooter = false;
reactFooterFailed = false;
+ protected get currentNoteId(): string | undefined {
+ return this.note?.id;
+ }
+
get shouldUseReactFooter(): boolean {
return this.useReactFooter && !this.reactFooterFailed;
}
@@ -627,7 +631,7 @@ export class NotebookParagraphComponent
this.handleKeyEvent(event.action, event.event);
this.notebookParagraphCodeEditorComponent?.handleKeyEvent(event.action);
});
- this.setResults(this.paragraph);
+ this.setParagraphSnapshot(this.paragraph);
this.originalText = this.paragraph.text;
this.isEntireNoteRunning =
this.noteStatusService.isEntireNoteRunning(this.note);
this.isParagraphRunning =
this.noteStatusService.isParagraphRunning(this.paragraph);
@@ -756,6 +760,9 @@ export class NotebookParagraphComponent
}
ngOnChanges(changes: SimpleChanges): void {
+ if (changes.paragraph || changes.note) {
+ this.setParagraphSnapshot(this.paragraph);
+ }
const { index, select, scrolled } = changes;
if (
(index && index.currentValue !== index.previousValue && this.select) ||
diff --git
a/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.spec.ts
b/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.spec.ts
new file mode 100644
index 0000000000..a9c9b75c40
--- /dev/null
+++
b/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.spec.ts
@@ -0,0 +1,143 @@
+/*
+ * 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 { QueryList } from '@angular/core';
+import { convertToParamMap } from '@angular/router';
+import { DatasetType, Note, ParagraphItem } from '@zeppelin/sdk';
+import { BehaviorSubject, EMPTY, of } from 'rxjs';
+import { afterEach, describe, expect, it, vi } from 'vitest';
+
+vi.mock('@zeppelin/services', () => ({
+ MessageService: class {},
+ HeliumService: class {},
+ NgZService: class {},
+ NoteStatusService: class {},
+ ReactFeatureService: class {}
+}));
+vi.mock('@zeppelin/core', async () => ({
+ ...(await import('../../../../core/paragraph-base/paragraph-base')),
+ ...(await import('../../../../core/message-listener/message-listener')),
+ publishedSymbol: Symbol('published')
+}));
+vi.mock('../../share/result/result.component', () => ({
NotebookParagraphResultComponent: class {} }));
+
+import { PublishedParagraphComponent } from './paragraph.component';
+
+const snapshot = (noteId: string, status: string, text: string): Note => ({
+ note: {
+ id: noteId,
+ name: noteId,
+ paragraphs: [
+ {
+ id: 'same-paragraph',
+ status,
+ text: '',
+ results: { msg: [{ type: DatasetType.TEXT, data: text }] },
+ config: {},
+ settings: { params: {}, forms: {} }
+ } as ParagraphItem
+ ]
+ } as Note['note']
+});
+
+const components: PublishedParagraphComponent[] = [];
+afterEach(() => components.splice(0).forEach(component =>
component.ngOnDestroy()));
+
+const setup = (useReact = false) => {
+ const params = new BehaviorSubject({ noteId: 'note-A', paragraphId:
'same-paragraph' });
+ const component = new PublishedParagraphComponent(
+ { receive: () => EMPTY, getNote: vi.fn() } as never,
+ { params, queryParamMap: of(convertToParamMap({})) } as never,
+ {} as never,
+ {} as never,
+ {} as never,
+ { isEnabled: () => useReact } as never,
+ { isParagraphRunning: (p: ParagraphItem) => p.status === 'RUNNING' } as
never,
+ {} as never,
+ { markForCheck: vi.fn() } as never
+ );
+ component.notebookParagraphResultComponents = new QueryList();
+ components.push(component);
+ return { component, params };
+};
+
+const append = (component: PublishedParagraphComponent, noteId: string, data:
string) =>
+ component.onParagraphAppendOutput({ noteId, paragraphId: 'same-paragraph',
index: 0, data });
+
+describe('Published paragraph authoritative snapshots', () => {
+ it('resumes from a same-note refreshed snapshot instead of cached output',
() => {
+ const { component } = setup();
+ component.getNote(snapshot('note-A', 'RUNNING', 'old'));
+ append(component, 'note-A', ' tail');
+ component.getNote(snapshot('note-A', 'RUNNING', 'fresh'));
+ append(component, 'note-A', ' next');
+ expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'fresh
next' }]);
+ });
+
+ it('unfreezes terminal output when the reused route loads a running
paragraph', () => {
+ const { component, params } = setup();
+ component.getNote(snapshot('note-A', 'FINISHED', 'A final'));
+ append(component, 'note-A', ' ignored');
+ params.next({ noteId: 'note-B', paragraphId: 'same-paragraph' });
+ component.getNote(snapshot('note-B', 'RUNNING', 'B first'));
+ append(component, 'note-B', ' second');
+ expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'B
first second' }]);
+ });
+
+ it('rejects stale NOTE and stream events while waiting for the new route
snapshot', () => {
+ const { component, params } = setup();
+ component.getNote(snapshot('note-A', 'RUNNING', 'A first'));
+ append(component, 'note-A', ' tail');
+ params.next({ noteId: 'note-B', paragraphId: 'same-paragraph' });
+ component.getNote(snapshot('note-A', 'RUNNING', 'stale A'));
+ append(component, 'note-A', ' stale tail');
+ append(component, 'note-B', ' before snapshot');
+ expect(component.results).toEqual([]);
+ component.getNote(snapshot('note-B', 'RUNNING', 'B first'));
+ append(component, 'note-A', ' foreign');
+ append(component, 'note-B', ' second');
+ expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'B
first second' }]);
+ });
+});
+
+describe('Published React streaming output', () => {
+ it.each(['append', 'update'])('refreshes React props after a streaming %s',
operation => {
+ const { component } = setup(true);
+ component.getNote(snapshot('note-A', 'RUNNING', 'initial'));
+ const initialProps = component.reactProps;
+ if (operation === 'append') {
+ append(component, 'note-A', ' tail');
+ } else {
+ component.onParagraphUpdateOutput({
+ noteId: 'note-A',
+ paragraphId: 'same-paragraph',
+ index: 0,
+ type: DatasetType.TEXT,
+ data: 'replacement'
+ });
+ }
+ const expected = [{ type: DatasetType.TEXT, data: operation === 'append' ?
'initial tail' : 'replacement' }];
+ expect(component.results).toEqual(expected);
+ expect(component.reactProps.results).toEqual(expected);
+ expect(component.reactProps).not.toBe(initialProps);
+ });
+
+ it('continues updating Angular results after React falls back', () => {
+ const { component } = setup(true);
+ component.reactFailed = true;
+ const updateResult = vi.fn();
+ component.notebookParagraphResultComponents.reset([{ updateResult } as
never]);
+ component.getNote(snapshot('note-A', 'RUNNING', 'initial'));
+ append(component, 'note-A', ' tail');
+ expect(updateResult).toHaveBeenCalledWith(expect.anything(), { type:
DatasetType.TEXT, data: 'initial tail' });
+ });
+});
diff --git
a/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.ts
b/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.ts
index 94f9af7d94..07d0439710 100644
---
a/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.ts
+++
b/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.ts
@@ -58,6 +58,10 @@ export class PublishedParagraphComponent extends
ParagraphBase implements Publis
@ViewChildren(NotebookParagraphResultComponent)
notebookParagraphResultComponents!:
QueryList<NotebookParagraphResultComponent>;
+ protected get currentNoteId(): string | null {
+ return this.noteId;
+ }
+
constructor(
public messageService: MessageService,
private activatedRoute: ActivatedRoute,
@@ -80,6 +84,7 @@ export class PublishedParagraphComponent extends
ParagraphBase implements Publis
}
this.noteId = params.noteId;
this.paragraphId = params.paragraphId!;
+ this.setParagraphSnapshot(undefined);
this.messageService.getNote(params.noteId);
});
}
@@ -87,21 +92,19 @@ export class PublishedParagraphComponent extends
ParagraphBase implements Publis
@MessageListener(OP.NOTE)
getNote(data: MessageReceiveDataTypeMap[OP.NOTE]) {
const note = data.note;
- if (!isNil(note)) {
- this.paragraph = note.paragraphs.find(p => p.id === this.paragraphId);
+ if (!isNil(note) && note.id === this.noteId) {
+ this.setParagraphSnapshot(note.paragraphs.find(p => p.id ===
this.paragraphId));
if (this.paragraph) {
if (!this.paragraph.results) {
this.showRunConfirmationModal();
}
if (this.useReact && !this.reactFailed) {
- this.setResults(this.paragraph);
this.reactProps = this.buildReactProps(this.paragraph);
this.isLoading = false;
this.cdr.markForCheck();
return;
}
- this.setResults(this.paragraph);
this.originalText = this.paragraph.text;
this.initializeDefault(this.paragraph.config, this.paragraph.settings);
} else {
@@ -153,8 +156,11 @@ export class PublishedParagraphComponent extends
ParagraphBase implements Publis
}
updateParagraphResult(resultIndex: number, config: ParagraphConfigResult,
result: ParagraphIResultsMsgItem): void {
- // In React mode the Angular result components never render, so this query
is empty.
- // The remote is refreshed from updateParagraphObjectWhenUpdated instead.
+ if (this.useReact && !this.reactFailed && this.paragraph) {
+ this.reactProps = this.buildReactProps(this.paragraph);
+ this.cdr.markForCheck();
+ return;
+ }
const resultComponent =
this.notebookParagraphResultComponents.toArray()[resultIndex];
if (resultComponent) {
resultComponent.updateResult(config, result);
