From 792a692d9b28968da955dd1a648183f08685779c Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 5 Sep 2026 16:44:26 +0900 Subject: [PATCH 01/11] [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 de86d67d90e77782e88bce7d1e1b2f3f0061c7c6 Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 5 Sep 2026 16:44:43 +0900 Subject: [PATCH 02/11] [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 33f20d6c169b9fe4a2607ada3ff52517efc56cbd Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 5 Sep 2026 16:45:54 +0900 Subject: [PATCH 03/11] [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 f81d79c1be6e2fec600434b4091161521fa04290 Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 5 Sep 2026 16:46:30 +0900 Subject: [PATCH 04/11] [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 8a937dc1a16..719b998356a 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 ca6b0062c30d6c3dcbbde7d60f9dfac1f841b3bc Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 5 Sep 2026 17:00:47 +0900 Subject: [PATCH 05/11] [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", From 7d8d886c3f02e2b87c2f508d7ebce9e01b3bc39a Mon Sep 17 00:00:00 2001 From: YONGJAE LEE Date: Wed, 9 Sep 2026 02:01:16 +0900 Subject: [PATCH 06/11] [ZEPPELIN-6659] Preserve output replacement and checkpoint ordering --- .../RemoteInterpreterEventServer.java | 19 ++- .../remote/AppendOutputRunner.java | 40 +++--- .../RemoteInterpreterEventServerTest.java | 128 ++++++++++++++++++ .../remote/AppendOutputRunnerTest.java | 107 ++++++++++++--- 4 files changed, 252 insertions(+), 42 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 fd9a4f329c0..c8f8886e6a9 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 @@ -231,6 +231,8 @@ public void updateOutput(OutputUpdateEvent event) throws InterpreterRPCException if (event.getAppId() == null) { runner.updateBuffer(event.getNoteId(), event.getParagraphId(), event.getIndex(), InterpreterResult.Type.valueOf(event.getType()), event.getData()); + // Complete replacements before the interpreter can publish its terminal result. + runner.run(); } else { appListener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), event.getIndex(), event.getAppId(), InterpreterResult.Type.valueOf(event.getType()), event.getData()); @@ -239,11 +241,15 @@ public void updateOutput(OutputUpdateEvent event) throws InterpreterRPCException @Override public void updateAllOutput(OutputUpdateAllEvent event) throws InterpreterRPCException, TException { - listener.onOutputClear(event.getNoteId(), event.getParagraphId()); - for (int i = 0; i < event.getMsg().size(); i++) { - RemoteInterpreterResultMessage msg = event.getMsg().get(i); - listener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), i, - InterpreterResult.Type.valueOf(msg.getType()), msg.getData()); + synchronized (runner) { + // Finish earlier output before the clear; keep replacements ahead of the next drain. + runner.run(); + listener.onOutputClear(event.getNoteId(), event.getParagraphId()); + for (int i = 0; i < event.getMsg().size(); i++) { + RemoteInterpreterResultMessage msg = event.getMsg().get(i); + listener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), i, + InterpreterResult.Type.valueOf(msg.getType()), msg.getData()); + } } } @@ -266,6 +272,9 @@ public void updateAppStatus(AppStatusUpdateEvent event) throws InterpreterRPCExc @Override public void checkpointOutput(String noteId, String paragraphId) throws InterpreterRPCException, TException { + // Drain replacements before checkpointing. + // Keep storage callbacks outside the runner lock to avoid blocking output delivery. + runner.run(); listener.checkpointOutput(noteId, paragraphId); } 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 55b20ad1e63..8bc063af106 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 @@ -47,35 +47,33 @@ public AppendOutputRunner(RemoteInterpreterProcessListener listener) { this.listener = listener; } + // Serialize scheduled and RPC drains to preserve callback order. + // Empty drains must return immediately to RPC callers. @Override - public void run() { + public synchronized void run() { Map stringBufferMap = new HashMap<>(); List 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); 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()); + 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; } @@ -112,8 +110,13 @@ private long flushAppendBuffers(Map stringBufferMap) { 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()); + 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; @@ -123,6 +126,7 @@ public void appendBuffer(String noteId, String paragraphId, int index, String ou 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/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java index ad385c612e7..db73b10a427 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.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 750119efa02..2d6e08beba7 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 @@ -30,12 +30,20 @@ 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; @@ -166,24 +174,85 @@ void testWarnLoggerForLargeData() throws InterruptedException { logger.addAppender(appender); runner.run(); - List 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 { @@ -253,4 +322,4 @@ private void loopForCompletingEvents(RemoteInterpreterProcessListener listener, } } } -} \ No newline at end of file +} From 6fc2828ed02cc4851c7c40381485691991d11567 Mon Sep 17 00:00:00 2001 From: YONGJAE LEE Date: Wed, 9 Sep 2026 02:01:25 +0900 Subject: [PATCH 07/11] [ZEPPELIN-6659] Preserve streaming state in notebook and published views --- .../e2e/models/published-paragraph-page.ts | 2 + .../paragraph/paragraph-functionality.spec.ts | 1 + .../published/published-streaming.spec.ts | 120 +++++++++ .../message-data-type-map.interface.spec.ts | 2 +- .../paragraph-base/paragraph-base.spec.ts | 247 ++++++++++++++++++ .../app/core/paragraph-base/paragraph-base.ts | 58 ++-- .../paragraph-output-state.spec.ts | 12 +- .../paragraph-base/paragraph-output-state.ts | 8 +- .../notebook/paragraph/paragraph.component.ts | 9 +- .../paragraph/paragraph.component.spec.ts | 143 ++++++++++ .../paragraph/paragraph.component.ts | 18 +- 11 files changed, 590 insertions(+), 30 deletions(-) create mode 100644 zeppelin-web-angular/e2e/tests/notebook/published/published-streaming.spec.ts create mode 100644 zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts create mode 100644 zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.spec.ts diff --git a/zeppelin-web-angular/e2e/models/published-paragraph-page.ts b/zeppelin-web-angular/e2e/models/published-paragraph-page.ts index e63350da022..8916fef0f7a 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 , 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/tests/notebook/paragraph/paragraph-functionality.spec.ts b/zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts index 719b998356a..b7f40798cf1 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 @@ -118,6 +118,7 @@ test.describe('Notebook Paragraph Functionality', () => { 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'); }); }); 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 00000000000..208a8721424 --- /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(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((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((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 index 000285ff98a..ec7797ecba1 100644 --- 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 @@ -27,7 +27,7 @@ it('declares the asymmetric paragraph output payloads sent by the server', () => ...append, type: DatasetType.TEXT }; - +expect(append).not.toHaveProperty('type'); + expect(append).not.toHaveProperty('type'); expect(update.type).toBe(DatasetType.TEXT); expectTypeOf().toEqualTypeOf(); expectTypeOf().not.toHaveProperty('type'); 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 00000000000..360801527e6 --- /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 tail' } + ]); + expect(component.updateParagraphResult).toHaveBeenCalledWith(2, expect.anything(), { + type: DatasetType.TEXT, + data: 'third tail' + }); + component.onParagraphAppendOutput({ noteId: 'note', paragraphId: 'A', index: 2, data: ' end' }); + expect(component.results[2].data).toBe('third tail 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 286f69eac6b..b9e514abd6e 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,7 +26,7 @@ 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'; @@ -73,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) @@ -121,25 +123,25 @@ export abstract class ParagraphBase extends MessageListenersManager { @MessageListener(OP.PARAGRAPH_APPEND_OUTPUT) onParagraphAppendOutput(data: MessageReceiveDataTypeMap[OP.PARAGRAPH_APPEND_OUTPUT]) { - if (data.paragraphId !== this.paragraph?.id) { + 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, result); + this.applyStreamingResult(data.index); } } @MessageListener(OP.PARAGRAPH_UPDATE_OUTPUT) onParagraphUpdateOutput(data: MessageReceiveDataTypeMap[OP.PARAGRAPH_UPDATE_OUTPUT]) { - if (data.paragraphId !== this.paragraph?.id) { + 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, result); + this.applyStreamingResult(data.index); } } @@ -147,17 +149,26 @@ export abstract class ParagraphBase extends MessageListenersManager { 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; - if (newRunActive && (!oldRunActive || newPara.dateStarted !== oldPara.dateStarted)) { + 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) { @@ -175,9 +186,6 @@ export abstract class ParagraphBase extends MessageListenersManager { }); this.cdr.markForCheck(); } - if (isTerminalParagraphStatus(newPara.status)) { - this.outputState.finish(newPara.results?.msg); - } } abstract updateParagraphResult( @@ -229,18 +237,23 @@ export abstract class ParagraphBase extends MessageListenersManager { } } - private applyStreamingResult(index: number, result: ParagraphIResultsMsgItem): void { + 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; - const config = this.paragraph.config.results?.[index] ?? { graph: new GraphConfig() }; - this.updateParagraphResult(index, config, result); + 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(); } @@ -279,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) || @@ -328,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; @@ -419,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 index 7bdc72e2281..a0c7a46d3c9 100644 --- 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 @@ -76,8 +76,8 @@ describe('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); + 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]); @@ -89,10 +89,10 @@ describe('ParagraphOutputState', () => { 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.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({ + expect(state.update(0, capturedType(update.data.type!), update.data.data!)).toEqual({ type: DatasetType.TEXT, data: 'first\nsecond\n' }); @@ -113,7 +113,7 @@ describe('ParagraphOutputState', () => { 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')!; + const finalAppend = [...events].reverse().find(event => event.op === 'PARAGRAPH_APPEND_OUTPUT')!; state.reset([{ type: DatasetType.TEXT, data: 'first\nsecond\n' }]); replay(state, [terminal, finalAppend]); 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 index a7e2720819d..d1c49604980 100644 --- 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 @@ -67,6 +67,12 @@ export class ParagraphOutputState { } snapshot(): ParagraphIResultsMsgItem[] { - return [...this.results]; + // 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/pages/workspace/notebook/paragraph/paragraph.component.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts index 17c9a4d5780..c7365dd3c9b 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 00000000000..a9c9b75c407 --- /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 94f9af7d943..07d0439710e 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; + 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); From c98cfab647c177f06abfafa9606aa51f139809b2 Mon Sep 17 00:00:00 2001 From: YONGJAE LEE Date: Thu, 10 Sep 2026 11:50:06 +0900 Subject: [PATCH 08/11] [ZEPPELIN-6659] Restrict unowned streaming output --- .../zeppelin/socket/NotebookServer.java | 22 +++-- .../NotebookServerStreamingScopeTest.java | 81 +++++++++++++++++++ 2 files changed, 96 insertions(+), 7 deletions(-) create mode 100644 zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerStreamingScopeTest.java 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 fcf76a8105d..4e21d2aca36 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 @@ -1769,7 +1769,19 @@ public void onOutputAppend(String noteId, String paragraphId, int index, String .put("paragraphId", paragraphId) .put("index", index) .put("data", output); - connectionManager.broadcast(noteId, msg); + try { + getNotebook().processNote(noteId, note -> { + if (note == null) { + LOGGER.warn("Note {} not found", noteId); + } else if (!note.isPersonalizedMode()) { + // Streaming events do not identify the user that owns the execution. + connectionManager.broadcast(noteId, msg); + } + return null; + }); + } catch (IOException e) { + LOGGER.warn("Fail to call onOutputAppend", e); + } } /** @@ -1798,12 +1810,8 @@ public void onOutputUpdated(String noteId, String paragraphId, int index, } 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 { + // Personalized output falls back to the user-specific terminal paragraph snapshot. + if (!note.isPersonalizedMode()) { connectionManager.broadcast(noteId, msg); } return null; 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 00000000000..1eda0b771fb --- /dev/null +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerStreamingScopeTest.java @@ -0,0 +1,81 @@ +/* + * 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.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 org.apache.zeppelin.conf.ZeppelinConfiguration; +import org.apache.zeppelin.interpreter.InterpreterResult; +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()); + } +} From 2fd233f2b9f931a7fae3a7b3552dd36ae1b608d6 Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 12 Sep 2026 19:20:40 +0900 Subject: [PATCH 09/11] [ZEPPELIN-6659] Keep unowned output out of the shared buffer in personalized mode Streaming UPDATE events carry no owner. In personalized mode they were still written to the shared Paragraph outputBuffer, which checkpointOutput saves as the shared result and which other users' paragraphs are cloned from, so one user's output could surface in another user's paragraph. Skip the buffer write together with the broadcast, and cover the checkpoint path with a regression test. --- .../zeppelin/socket/NotebookServer.java | 13 ++++++++----- .../NotebookServerStreamingScopeTest.java | 19 +++++++++++++++++++ 2 files changed, 27 insertions(+), 5 deletions(-) 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 4e21d2aca36..0681042ce9e 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 @@ -1808,12 +1808,15 @@ public void onOutputUpdated(String noteId, String paragraphId, int index, LOGGER.warn("Note {} not found", noteId); return null; } - Paragraph paragraph = note.getParagraph(paragraphId); - paragraph.updateOutputBuffer(index, type, output); - // Personalized output falls back to the user-specific terminal paragraph snapshot. - if (!note.isPersonalizedMode()) { - connectionManager.broadcast(noteId, msg); + if (note.isPersonalizedMode()) { + // 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) { 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 index 1eda0b771fb..5ba1a871bf5 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerStreamingScopeTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerStreamingScopeTest.java @@ -16,6 +16,7 @@ */ package org.apache.zeppelin.socket; +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; @@ -25,8 +26,11 @@ 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; @@ -78,4 +82,19 @@ void personalizedNoteDoesNotReceiveUnownedIncrementalOutput() { 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 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); + } } From 3876b71fc3239637e5c0ebf5851e350334883673 Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 12 Sep 2026 19:20:40 +0900 Subject: [PATCH 10/11] [ZEPPELIN-6659] Register streaming output in the notebook parity registry Add NB-PARITY-022 for streaming interpreter output and tag the existing browser test so the registry records it as covered. --- .../e2e/scenarios/notebook-parity.json | 80 +++++++++++++++++++ .../e2e/scenarios/notebook-parity.md | 13 +++ .../paragraph/paragraph-functionality.spec.ts | 56 +++++++------ 3 files changed, 123 insertions(+), 26 deletions(-) diff --git a/zeppelin-web-angular/e2e/scenarios/notebook-parity.json b/zeppelin-web-angular/e2e/scenarios/notebook-parity.json index 9ee808f9078..7da8c388176 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 3da42b6e890..e90d35f5b8e 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
@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
@NB-PARITY-011 | | | NB-PARITY-021 | result | Text and table result displays preserve output semantics after paragraph execution | partial | owner: allow
writer: allow
reader: deny
runner: allow | zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts
@NB-PARITY-021 | ZEPPELIN-6514, ZEPPELIN-6516 | +| NB-PARITY-022 | result | Streaming interpreter output accumulates while a paragraph is running | covered | owner: allow
writer: allow
reader: deny
runner: allow | zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts
@NB-PARITY-022 | | | NB-PARITY-050 | persistence | Notebook editor persists the latest text after typing stops | gap | owner: allow
writer: allow
reader: deny
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
writer: allow
reader: deny
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 accessible table output row by row so migrated React rendering can be compared without relying on screenshots alone. +### 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 b7f40798cf1..9211fd5e743 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 @@ -95,32 +95,36 @@ 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'); - await expect(paragraphPage.resultDisplay).toHaveText('first\nsecond\nthird\n'); - }); - }); + 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'); From 501ea132916bc3c6d36b09f4ddfadae153f32d17 Mon Sep 17 00:00:00 2001 From: miinhho Date: Sun, 13 Sep 2026 01:47:45 +0900 Subject: [PATCH 11/11] [ZEPPELIN-6659] Fix personalized clear and stale pending output --- .../zeppelin/socket/NotebookServer.java | 12 ++++--- .../NotebookServerStreamingScopeTest.java | 31 +++++++++++++++++++ .../paragraph-base/paragraph-base.spec.ts | 6 ++-- .../paragraph-output-state.spec.ts | 17 ++++++++++ .../paragraph-base/paragraph-output-state.ts | 3 +- 5 files changed, 61 insertions(+), 8 deletions(-) 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 0681042ce9e..28fde11fc71 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 @@ -1839,11 +1839,15 @@ public void onOutputClear(String noteId, String paragraphId) { if (note == null) { // It is possible the note is removed, but the job is still running LOGGER.warn("Note {} doesn't existed, it maybe deleted.", noteId); - } else { - note.clearParagraphOutput(paragraphId); - Paragraph paragraph = note.getParagraph(paragraphId); - broadcastParagraph(note, paragraph, MSG_ID_NOT_DEFINED); + return null; + } + if (note.isPersonalizedMode()) { + // Streaming events carry no owner, so they must not mutate shared paragraph state. + return null; } + note.clearParagraphOutput(paragraphId); + Paragraph paragraph = note.getParagraph(paragraphId); + broadcastParagraph(note, paragraph, MSG_ID_NOT_DEFINED); return null; }); 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 index 5ba1a871bf5..cdc4d5c7bf3 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerStreamingScopeTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerStreamingScopeTest.java @@ -16,6 +16,8 @@ */ 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; @@ -97,4 +99,33 @@ void personalizedCheckpointDoesNotExposeUnownedOutputToOtherUsers() { 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/src/app/core/paragraph-base/paragraph-base.spec.ts b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts index 360801527e6..34d5cc44183 100644 --- 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 @@ -234,14 +234,14 @@ describe('ParagraphBase streaming boundaries', () => { expect(component.results).toEqual([ { type: DatasetType.TEXT, data: 'first\n' }, { type: DatasetType.TABLE, data: 'second' }, - { type: DatasetType.TEXT, data: 'third tail' } + { type: DatasetType.TEXT, data: 'third' } ]); expect(component.updateParagraphResult).toHaveBeenCalledWith(2, expect.anything(), { type: DatasetType.TEXT, - data: 'third tail' + data: 'third' }); component.onParagraphAppendOutput({ noteId: 'note', paragraphId: 'A', index: 2, data: ' end' }); - expect(component.results[2].data).toBe('third tail end'); + expect(component.results[2].data).toBe('third end'); component.ngOnDestroy(); }); }); 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 index a0c7a46d3c9..e9ce623d4db 100644 --- 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 @@ -98,6 +98,23 @@ describe('ParagraphOutputState', () => { }); }); + 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' }]); 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 index d1c49604980..93934ededca 100644 --- 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 @@ -38,9 +38,10 @@ export class ParagraphOutputState { 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: data === '' ? (this.pendingAppends.get(index) ?? '') : data }; this.pendingAppends.delete(index); this.results[index] = result;