From 047c88b422f8892a1da6b11ac9c8d969b9de09f1 Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 5 Sep 2026 16:44:26 +0900 Subject: [PATCH 1/5] [ZEPPELIN-6659] Define paragraph streaming output payloads --- .../message-data-type-map.interface.spec.ts | 38 +++++++++++++++++++ .../message-data-type-map.interface.ts | 4 ++ .../interfaces/message-paragraph.interface.ts | 11 ++++++ 3 files changed, 53 insertions(+) create mode 100644 zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.spec.ts 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 00000000000..000285ff98a --- /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().toEqualTypeOf(); + expectTypeOf().not.toHaveProperty('type'); + expectTypeOf().toEqualTypeOf(); + expectTypeOf() + .toHaveProperty('type') + .toEqualTypeOf(); +}); 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 f86dbdb2b12..6c023a09231 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 f75cd1f5f31..f9e01351e2e 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; From c1213a85e5bef9382db4efd38fe4d56d09964c77 Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 5 Sep 2026 16:44:43 +0900 Subject: [PATCH 2/5] [ZEPPELIN-6659] Preserve paragraph output event ordering --- .../RemoteInterpreterEventServer.java | 2 +- .../remote/AppendOutputRunner.java | 41 +++++++++++++------ .../remote/UpdateOutputBuffer.java | 41 +++++++++++++++++++ .../remote/AppendOutputRunnerTest.java | 21 ++++++++++ 4 files changed, 92 insertions(+), 13 deletions(-) create mode 100644 zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/UpdateOutputBuffer.java 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 657ad593c8b..fd9a4f329c0 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,7 +229,7 @@ public void appendOutput(OutputAppendEvent event) throws InterpreterRPCException @Override public void updateOutput(OutputUpdateEvent event) throws InterpreterRPCException, TException { if (event.getAppId() == null) { - listener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), event.getIndex(), + runner.updateBuffer(event.getNoteId(), event.getParagraphId(), event.getIndex(), InterpreterResult.Type.valueOf(event.getType()), event.getData()); } else { appListener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), event.getIndex(), 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 93dd3282969..55b20ad1e63 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.LinkedBlockingQueue; /** - * This thread sends paragraph's append-data - * periodically, rather than continously, with - * a period of BUFFER_TIME_MS. It handles append-data - * for all paragraphs across all notebooks. + * Sends paragraph output periodically. Adjacent append events are batched, while update events + * share the same queue so that they cannot overtake earlier appends. */ public class AppendOutputRunner implements Runnable { @@ -70,7 +69,16 @@ public void run() { Long processingStartTime = System.currentTimeMillis(); queue.drainTo(list); - for (AppendOutputBuffer buffer: list) { + Long sizeProcessed = Long.valueOf(0); + for (AppendOutputBuffer buffer : list) { + if (buffer instanceof UpdateOutputBuffer) { + sizeProcessed += flushAppendBuffers(stringBufferMap); + UpdateOutputBuffer update = (UpdateOutputBuffer) buffer; + listener.onOutputUpdated(update.getNoteId(), update.getParagraphId(), update.getIndex(), + update.getType(), update.getData()); + continue; + } + String noteId = buffer.getNoteId(); String paragraphId = buffer.getParagraphId(); int index = buffer.getIndex(); @@ -82,6 +90,7 @@ public void run() { builder.append(buffer.getData()); stringBufferMap.put(stringBufferKey, builder); } + sizeProcessed += flushAppendBuffers(stringBufferMap); Long processingTime = System.currentTimeMillis() - processingStartTime; if (processingTime > SAFE_PROCESSING_TIME) { @@ -90,7 +99,15 @@ public void run() { LOGGER.debug("Processing time for append-output took {} milliseconds", processingTime); } - Long sizeProcessed = Long.valueOf(0); + if (sizeProcessed > SAFE_PROCESSING_STRING_SIZE) { + LOGGER.warn("Processing size for buffered append-output is high: {} characters.", sizeProcessed); + } else { + LOGGER.debug("Processing size for append-output is {} characters", sizeProcessed); + } + } + + private long flushAppendBuffers(Map stringBufferMap) { + long sizeProcessed = 0; for (Entry stringBufferMapEntry : stringBufferMap.entrySet()) { String stringBufferKey = stringBufferMapEntry.getKey(); StringBuilder buffer = stringBufferMapEntry.getValue(); @@ -98,16 +115,16 @@ public void run() { String[] keys = stringBufferKey.split(":"); listener.onOutputAppend(keys[0], keys[1], Integer.parseInt(keys[2]), buffer.toString()); } - - if (sizeProcessed > SAFE_PROCESSING_STRING_SIZE) { - LOGGER.warn("Processing size for buffered append-output is high: {} characters.", sizeProcessed); - } else { - LOGGER.debug("Processing size for append-output is {} characters", sizeProcessed); - } + stringBufferMap.clear(); + return sizeProcessed; } public void appendBuffer(String noteId, String paragraphId, int index, String outputToAppend) { queue.offer(new AppendOutputBuffer(noteId, paragraphId, index, outputToAppend)); } + public void updateBuffer(String noteId, String paragraphId, int index, + InterpreterResult.Type type, String output) { + queue.offer(new UpdateOutputBuffer(noteId, paragraphId, index, type, output)); + } } 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 00000000000..15de2d5092b --- /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/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java index 1d8d273cb73..750119efa02 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,12 +17,14 @@ package org.apache.zeppelin.interpreter.remote; +import org.apache.zeppelin.interpreter.InterpreterResult; import org.apache.log4j.AppenderSkeleton; import org.apache.log4j.Level; import org.apache.log4j.Logger; import org.apache.log4j.spi.LoggingEvent; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; +import org.mockito.InOrder; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; @@ -39,6 +41,7 @@ import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.Mockito.atMost; import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -88,6 +91,24 @@ public void testMultipleEventsOfSameParagraph() throws InterruptedException { verify(listener, times(1)).onOutputAppend(note1, para1, 0, "data1\ndata2\ndata3\n"); } + @Test + void testUpdateDoesNotOvertakeQueuedAppend() { + RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); + AppendOutputRunner runner = new AppendOutputRunner(listener); + runner.appendBuffer("note", "para", 0, "before-1\n"); + runner.appendBuffer("note", "para", 0, "before-2\n"); + runner.updateBuffer("note", "para", 0, InterpreterResult.Type.TEXT, "replacement\n"); + runner.appendBuffer("note", "para", 0, "after\n"); + + runner.run(); + + InOrder order = inOrder(listener); + order.verify(listener).onOutputAppend("note", "para", 0, "before-1\nbefore-2\n"); + order.verify(listener).onOutputUpdated( + "note", "para", 0, InterpreterResult.Type.TEXT, "replacement\n"); + order.verify(listener).onOutputAppend("note", "para", 0, "after\n"); + } + @Test void testMultipleEventsOfDifferentParagraphs() throws InterruptedException { RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class); From 1a2161721a92c5bbe53140278625d53465ee07ab Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 5 Sep 2026 16:45:54 +0900 Subject: [PATCH 3/5] [ZEPPELIN-6659] Render streaming paragraph output in the New UI --- .../app/core/paragraph-base/paragraph-base.ts | 58 +++++++ .../paragraph-output-state.spec.ts | 134 ++++++++++++++++ .../paragraph-base/paragraph-output-state.ts | 72 +++++++++ .../paragraph-output-stream.capture.json | 149 ++++++++++++++++++ 4 files changed, 413 insertions(+) create mode 100644 zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.spec.ts create mode 100644 zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.ts create mode 100644 zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-stream.capture.json 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 4e7c0e0fde8..286f69eac6b 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 @@ -32,6 +32,7 @@ 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, @@ -114,6 +119,30 @@ export abstract class ParagraphBase extends MessageListenersManager { } } + @MessageListener(OP.PARAGRAPH_APPEND_OUTPUT) + onParagraphAppendOutput(data: MessageReceiveDataTypeMap[OP.PARAGRAPH_APPEND_OUTPUT]) { + if (data.paragraphId !== this.paragraph?.id) { + return; + } + this.initializeOutputState(); + const result = this.outputState.append(data.index, data.data); + if (result) { + this.applyStreamingResult(data.index, result); + } + } + + @MessageListener(OP.PARAGRAPH_UPDATE_OUTPUT) + onParagraphUpdateOutput(data: MessageReceiveDataTypeMap[OP.PARAGRAPH_UPDATE_OUTPUT]) { + if (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, result); + } + } + @MessageListener(OP.PARAGRAPH) paragraphData(data: MessageReceiveDataTypeMap[OP.PARAGRAPH]) { const oldPara = this.paragraph; @@ -124,6 +153,11 @@ export abstract class ParagraphBase extends MessageListenersManager { 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; + if (newRunActive && (!oldRunActive || newPara.dateStarted !== oldPara.dateStarted)) { + this.outputState.reset(); + } if (this.isUpdateRequired(oldPara, newPara)) { this.updateParagraph(oldPara, newPara, () => { if (newPara.results && newPara.results.msg) { @@ -141,6 +175,9 @@ export abstract class ParagraphBase extends MessageListenersManager { }); this.cdr.markForCheck(); } + if (isTerminalParagraphStatus(newPara.status)) { + this.outputState.finish(newPara.results?.msg); + } } abstract updateParagraphResult( @@ -186,6 +223,27 @@ 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, result: ParagraphIResultsMsgItem): void { + if (!this.paragraph) { + return; + } + const results = this.outputState.snapshot(); + if (!this.paragraph.results) { + this.paragraph.results = {}; + } + this.paragraph.results.msg = results; + this.results = results; + const config = this.paragraph.config.results?.[index] ?? { graph: new GraphConfig() }; + this.updateParagraphResult(index, config, result); + this.cdr.markForCheck(); + } + updateParagraph(oldPara: ParagraphItem, newPara: ParagraphItem, updateCallback: () => void) { // 1. can't update on revision view if (!this.revisionView) { 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 00000000000..7bdc72e2281 --- /dev/null +++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.spec.ts @@ -0,0 +1,134 @@ +/* + * 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('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.findLast(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 00000000000..a7e2720819d --- /dev/null +++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.ts @@ -0,0 +1,72 @@ +/* + * 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(); + 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; + } + + const result = { + type, + data: data + (this.pendingAppends.get(index) ?? '') + }; + 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[] { + return [...this.results]; + } +} 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 00000000000..544d2b0034f --- /dev/null +++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-stream.capture.json @@ -0,0 +1,149 @@ +{ + "_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", + "gitCommitId": "c1213a85e5bef9382db4efd38fe4d56d09964c77", + "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": "", + "paragraphId": "", + "index": 0, + "type": "TEXT", + "data": "" + } + }, + { + "op": "PARAGRAPH_APPEND_OUTPUT", + "data": { + "noteId": "", + "paragraphId": "", + "index": 0, + "data": "first\n" + } + }, + { + "op": "PARAGRAPH_APPEND_OUTPUT", + "data": { + "noteId": "", + "paragraphId": "", + "index": 0, + "data": "second\n" + } + }, + { + "op": "PARAGRAPH_APPEND_OUTPUT", + "data": { + "noteId": "", + "paragraphId": "", + "index": 0, + "data": "third\n" + } + } + ] + }, + "enabled": { + "configuration": { + "zeppelin.websocket.paragraph_status_progress.enable": true + }, + "events": [ + { + "elapsedMs": 0, + "op": "PARAGRAPH_UPDATE_OUTPUT", + "data": { + "noteId": "", + "paragraphId": "", + "index": 0, + "type": "TEXT", + "data": "" + } + }, + { + "elapsedMs": 107, + "op": "PARAGRAPH_APPEND_OUTPUT", + "data": { + "noteId": "", + "paragraphId": "", + "index": 0, + "data": "first\n" + } + }, + { + "elapsedMs": 2984, + "op": "PARAGRAPH_APPEND_OUTPUT", + "data": { + "noteId": "", + "paragraphId": "", + "index": 0, + "data": "second\n" + } + }, + { + "elapsedMs": 7990, + "op": "PARAGRAPH_APPEND_OUTPUT", + "data": { + "noteId": "", + "paragraphId": "", + "index": 0, + "data": "third\n" + } + }, + { + "elapsedMs": 8027, + "op": "PARAGRAPH", + "data": { + "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": "", + "status": "FINISHED", + "results": { + "code": "SUCCESS", + "msg": [ + { + "type": "TEXT", + "data": "first\nsecond\nthird\n" + } + ] + } + } + } + } + ] + } +} From 39416a573f670b3aeb72a4a8ca3931877f312f4f Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 5 Sep 2026 16:46:30 +0900 Subject: [PATCH 4/5] [ZEPPELIN-6659] Cover incremental paragraph output in E2E --- .../paragraph/paragraph-functionality.spec.ts | 31 +++++++++++++++++-- 1 file changed, 29 insertions(+), 2 deletions(-) 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 105c4226202..02da92141af 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,32 @@ test.describe('Notebook Paragraph Functionality', () => { await expect(paragraphPage.resultDisplay).not.toBeEmpty(); }); + test('should accumulate interpreter output while the paragraph is running', async ({ page }) => { + await test.step('Given a shell paragraph that emits three delayed output chunks', async () => { + await setParagraphText( + page, + testNotebook.noteId, + testNotebook.paragraphId, + '%sh\necho first; sleep 3; echo second; sleep 5; echo third' + ); + await page.reload(); + await expect(paragraphPage.paragraphContainer).toBeVisible({ timeout: 30000 }); + }); + + await test.step('When the paragraph runs', async () => { + await paragraphPage.runParagraph(); + }); + + await test.step('Then output accumulates before the paragraph finishes', async () => { + await expect(paragraphPage.resultDisplay).toContainText('first', { timeout: 30000 }); + await expect(paragraphPage.status).toHaveText('RUNNING'); + await expect(paragraphPage.resultDisplay).toContainText(/first\s+second/, { timeout: 10000 }); + await expect(paragraphPage.status).toHaveText('RUNNING'); + await expect(paragraphPage.resultDisplay).toContainText(/first\s+second\s+third/, { timeout: 10000 }); + await expect(paragraphPage.status).toHaveText('FINISHED'); + }); + }); + test('should display dynamic forms', async ({ page }) => { test.skip(!!process.env.CI, 'Dynamic form tests require a Spark interpreter — skipped on CI'); From 953457739de355bd22c0746163d262086819d25a Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 5 Sep 2026 17:00:47 +0900 Subject: [PATCH 5/5] [ZEPPELIN-6659] Remove unstable capture commit metadata --- .../app/core/paragraph-base/paragraph-output-stream.capture.json | 1 - 1 file changed, 1 deletion(-) 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 index 544d2b0034f..655ca5fbba3 100644 --- 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 @@ -4,7 +4,6 @@ "capture": { "issue": "ZEPPELIN-6659", "zeppelinVersion": "0.13.0-SNAPSHOT", - "gitCommitId": "c1213a85e5bef9382db4efd38fe4d56d09964c77", "capturedAt": "2026-09-05T07:15:34Z", "browser": "Chromium", "interpreter": "sh",