From 61b91dbea040604ecf1557a6ca30ff352e4e00f1 Mon Sep 17 00:00:00 2001 From: gyowoo1113 Date: Wed, 16 Sep 2026 14:41:34 +0900 Subject: [PATCH 01/15] [ZEPPELIN-6683] Add envelope-aware WebSocket receive path --- .../projects/zeppelin-sdk/src/message.ts | 39 ++++++++++++------- .../core/message-listener/message-listener.ts | 27 +++++++++---- 2 files changed, 45 insertions(+), 21 deletions(-) diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts index 4d559a86aa1..f9dff2e804a 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts @@ -175,21 +175,15 @@ export class Message { } receive(op: K): Observable[K]> { - const guard = getMessagePayloadGuard(op); - - return this.received$.pipe( - filter(message => message.op === op), - filter(message => { - if (!guard || guard(message.data)) { - return true; - } + return this.receiveMessage(op).pipe(map(message => message.data)) as Observable< + Record[K] + >; + } - // The payload can be large and carries note names, so log the OP alone. - console.warn(`Dropped WebSocket OP ${String(op)}: payload failed validation`); - return false; - }), - map(message => message.data) - ) as Observable[K]>; + receiveEnvelope( + op: K + ): Observable> { + return this.receiveMessage(op) as Observable>; } shortCircuit(message: WebSocketMessage) { @@ -565,4 +559,21 @@ export class Message { formName }); } + + private receiveMessage(op: K) { + const guard = getMessagePayloadGuard(op); + + return this.received$.pipe( + filter(message => message.op === op), + filter(message => { + if (!guard || guard(message.data)) { + return true; + } + + // The payload can be large and carries note names, so log the OP alone. + console.warn(`Dropped WebSocket OP ${String(op)}: payload failed validation`); + return false; + }) + ); + } } diff --git a/zeppelin-web-angular/src/app/core/message-listener/message-listener.ts b/zeppelin-web-angular/src/app/core/message-listener/message-listener.ts index 1b2f0209ae7..93c6294700c 100644 --- a/zeppelin-web-angular/src/app/core/message-listener/message-listener.ts +++ b/zeppelin-web-angular/src/app/core/message-listener/message-listener.ts @@ -11,9 +11,9 @@ */ import { Component, OnDestroy } from '@angular/core'; -import { Subscriber } from 'rxjs'; +import { Observable, Subscriber } from 'rxjs'; -import { Message, MessageReceiveDataTypeMap, ReceiveArgumentsType } from '@zeppelin/sdk'; +import { Message, MessageReceiveDataTypeMap } from '@zeppelin/sdk'; @Component({ template: '', @@ -34,13 +34,18 @@ export class MessageListenersManager implements OnDestroy { } } -export function MessageListener(op: K) { +type ListenerArgumentsType = T extends undefined ? () => void : (data: T) => void; + +const createMessageListener = ( + op: K, + receiver: (messageService: Message, op: K) => Observable +) => { return function ( target: MessageListenersManager, propertyKey: string, - descriptor: TypedPropertyDescriptor> + descriptor: TypedPropertyDescriptor> ) { - const oldValue = descriptor.value as ReceiveArgumentsType; + const oldValue = descriptor.value as ListenerArgumentsType; const fn = function (this: MessageListenersManager) { if (!this.__zeppelinMessageListeners$__) { @@ -48,7 +53,7 @@ export function MessageListener(op: K } this.__zeppelinMessageListeners$__.add( - this.messageService.receive(op).subscribe(data => { + receiver(this.messageService, op).subscribe(data => { try { // @ts-ignore oldValue.apply(this, [data]); @@ -68,4 +73,12 @@ export function MessageListener(op: K return descriptor; }; -} +}; + +export const MessageListener = (op: K) => { + return createMessageListener(op, (messageService, targetOp) => messageService.receive(targetOp)); +}; + +export const MessageEnvelopeListener = (op: K) => { + return createMessageListener(op, (messageService, targetOp) => messageService.receiveEnvelope(targetOp)); +}; From 98a1882af666e657af75571fafedacae4e0f04be Mon Sep 17 00:00:00 2001 From: gyowoo1113 Date: Wed, 16 Sep 2026 16:31:17 +0900 Subject: [PATCH 02/15] [ZEPPELIN-6683] Expose WebSocket reply envelopes to notebook listeners --- .../workspace/notebook/notebook.component.ts | 27 +++++++++++++------ 1 file changed, 19 insertions(+), 8 deletions(-) diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts index 3dd6e739607..f0398cbae8e 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts @@ -28,7 +28,7 @@ import { distinctUntilChanged, distinctUntilKeyChanged, startWith, takeUntil } f import { NzResizeEvent } from 'ng-zorro-antd/resizable'; -import { MessageListener, MessageListenersManager } from '@zeppelin/core'; +import { MessageEnvelopeListener, MessageListener, MessageListenersManager } from '@zeppelin/core'; import { Permissions } from '@zeppelin/interfaces'; import { DynamicFormParams, @@ -36,7 +36,8 @@ import { MessageReceiveDataTypeMap, Note, OP, - RevisionListItem + RevisionListItem, + WebSocketMessage } from '@zeppelin/sdk'; import { MessageService, @@ -116,8 +117,13 @@ export class NotebookComponent extends MessageListenersManager implements OnInit } } - @MessageListener(OP.INTERPRETER_BINDINGS) - loadInterpreterBindings(data: MessageReceiveDataTypeMap[OP.INTERPRETER_BINDINGS]) { + @MessageEnvelopeListener(OP.INTERPRETER_BINDINGS) + loadInterpreterBindings(message: WebSocketMessage) { + const data = message.data; + if (data === undefined) { + return; + } + this.interpreterBindings = data.interpreterBindings; if (!this.interpreterBindings.some(item => item.selected)) { this.activatedExtension = 'interpreter'; @@ -197,8 +203,8 @@ export class NotebookComponent extends MessageListenersManager implements OnInit } } - @MessageListener(OP.SET_NOTE_REVISION) - setNoteRevision(_data: MessageReceiveDataTypeMap[OP.SET_NOTE_REVISION]) { + @MessageEnvelopeListener(OP.SET_NOTE_REVISION) + setNoteRevision(_message: WebSocketMessage) { const { noteId } = this.activatedRoute.snapshot.params; this.router.navigate(['/notebook', noteId]).then(); } @@ -255,8 +261,13 @@ export class NotebookComponent extends MessageListenersManager implements OnInit this.cdr.markForCheck(); } - @MessageListener(OP.LIST_REVISION_HISTORY) - listRevisionHistory(data: MessageReceiveDataTypeMap[OP.LIST_REVISION_HISTORY]) { + @MessageEnvelopeListener(OP.LIST_REVISION_HISTORY) + listRevisionHistory(message: WebSocketMessage) { + const data = message.data; + if (data === undefined) { + return; + } + this.noteRevisions = data.revisionList; if (this.noteRevisions) { if (this.noteRevisions.length === 0 || this.noteRevisions[0].id !== 'Head') { From 6c46f750769c40f9a200be6b42f056e00bd226b2 Mon Sep 17 00:00:00 2001 From: gyowoo1113 Date: Wed, 16 Sep 2026 16:32:18 +0900 Subject: [PATCH 03/15] [ZEPPELIN-6683] Track note context for pending WebSocket requests --- .../src/app/services/message.service.ts | 41 ++++++++++++------- 1 file changed, 27 insertions(+), 14 deletions(-) diff --git a/zeppelin-web-angular/src/app/services/message.service.ts b/zeppelin-web-angular/src/app/services/message.service.ts index 050d78c44cc..fc3e2f0ed31 100644 --- a/zeppelin-web-angular/src/app/services/message.service.ts +++ b/zeppelin-web-angular/src/app/services/message.service.ts @@ -41,6 +41,7 @@ import { TicketService } from './ticket.service'; }) export class MessageService extends Message implements OnDestroy { private readonly localAddFocusMsgIds = new Set(); + private readonly pendingNoteRequests = new Map(); constructor( private baseUrlService: BaseUrlService, @@ -93,15 +94,11 @@ export class MessageService extends Message implements OnDestroy { return this.localAddFocusMsgIds.delete(msgId); } - private captureLocalAddFocusMsgId(sendMessage: () => void): void { - const subscription = super - .sent() - .pipe(take(1)) - .subscribe(message => { - if (message.msgId) { - this.localAddFocusMsgIds.add(message.msgId); - } - }); + private captureSentMessage( + sendMessage: () => void, + onSent: (message: WebSocketMessage) => void + ): void { + const subscription = super.sent().pipe(take(1)).subscribe(onSent); try { sendMessage(); } catch (error) { @@ -110,6 +107,22 @@ export class MessageService extends Message implements OnDestroy { } } + private captureLocalAddFocusMsgId(sendMessage: () => void): void { + this.captureSentMessage(sendMessage, message => { + if (message.msgId) { + this.localAddFocusMsgIds.add(message.msgId); + } + }); + } + + private capturePendingNoteRequest(noteId: string, sendMessage: () => void): void { + this.captureSentMessage(sendMessage, message => { + if (message.msgId) { + this.pendingNoteRequests.set(message.msgId, noteId); + } + }); + } + opened(): Observable { return super.opened(); } @@ -317,15 +330,15 @@ export class MessageService extends Message implements OnDestroy { } checkpointNote(noteId: string, commitMessage: string): void { - super.checkpointNote(noteId, commitMessage); + this.capturePendingNoteRequest(noteId, () => super.checkpointNote(noteId, commitMessage)); } setNoteRevision(noteId: string, revisionId: string): void { - super.setNoteRevision(noteId, revisionId); + this.capturePendingNoteRequest(noteId, () => super.setNoteRevision(noteId, revisionId)); } listRevisionHistory(noteId: string): void { - super.listRevisionHistory(noteId); + this.capturePendingNoteRequest(noteId, () => super.listRevisionHistory(noteId)); } noteRevision(noteId: string, revisionId: string): void { @@ -349,11 +362,11 @@ export class MessageService extends Message implements OnDestroy { } getInterpreterBindings(noteId: string): void { - super.getInterpreterBindings(noteId); + this.capturePendingNoteRequest(noteId, () => super.getInterpreterBindings(noteId)); } saveInterpreterBindings(noteId: string, selectedSettingIds: string[]): void { - super.saveInterpreterBindings(noteId, selectedSettingIds); + this.capturePendingNoteRequest(noteId, () => super.saveInterpreterBindings(noteId, selectedSettingIds)); } getInterpreterSettings(): void { From 5d1e6ebdb74bcef3b5087a2b9f4ec1467d6010e9 Mon Sep 17 00:00:00 2001 From: gyowoo1113 Date: Wed, 16 Sep 2026 16:53:53 +0900 Subject: [PATCH 04/15] [ZEPPELIN-6683] Add client-side stale reply correlation --- .../workspace/notebook/notebook.component.ts | 19 ++++++++++++++++++- .../src/app/services/message.service.ts | 15 +++++++++++++++ 2 files changed, 33 insertions(+), 1 deletion(-) diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts index f0398cbae8e..2af9fca28e2 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts @@ -119,6 +119,12 @@ export class NotebookComponent extends MessageListenersManager implements OnInit @MessageEnvelopeListener(OP.INTERPRETER_BINDINGS) loadInterpreterBindings(message: WebSocketMessage) { + const { noteId } = this.activatedRoute.snapshot.params; + + if (!this.messageService.consumePendingNoteRequest(message.msgId, noteId)) { + return; + } + const data = message.data; if (data === undefined) { return; @@ -204,8 +210,13 @@ export class NotebookComponent extends MessageListenersManager implements OnInit } @MessageEnvelopeListener(OP.SET_NOTE_REVISION) - setNoteRevision(_message: WebSocketMessage) { + setNoteRevision(message: WebSocketMessage) { const { noteId } = this.activatedRoute.snapshot.params; + + if (!this.messageService.consumePendingNoteRequest(message.msgId, noteId)) { + return; + } + this.router.navigate(['/notebook', noteId]).then(); } @@ -263,6 +274,12 @@ export class NotebookComponent extends MessageListenersManager implements OnInit @MessageEnvelopeListener(OP.LIST_REVISION_HISTORY) listRevisionHistory(message: WebSocketMessage) { + const { noteId } = this.activatedRoute.snapshot.params; + + if (!this.messageService.consumePendingNoteRequest(message.msgId, noteId)) { + return; + } + const data = message.data; if (data === undefined) { return; diff --git a/zeppelin-web-angular/src/app/services/message.service.ts b/zeppelin-web-angular/src/app/services/message.service.ts index fc3e2f0ed31..3ceac51ec84 100644 --- a/zeppelin-web-angular/src/app/services/message.service.ts +++ b/zeppelin-web-angular/src/app/services/message.service.ts @@ -94,6 +94,21 @@ export class MessageService extends Message implements OnDestroy { return this.localAddFocusMsgIds.delete(msgId); } + consumePendingNoteRequest(msgId: string | undefined, activeNoteId: string): boolean { + if (!msgId) { + return false; + } + + const requestedNoteId = this.pendingNoteRequests.get(msgId); + if (!requestedNoteId) { + return false; + } + + this.pendingNoteRequests.delete(msgId); + + return requestedNoteId === activeNoteId; + } + private captureSentMessage( sendMessage: () => void, onSent: (message: WebSocketMessage) => void From db48085fe0421d7e8378faa551e896791e442efb Mon Sep 17 00:00:00 2001 From: gyowoo1113 Date: Wed, 16 Sep 2026 17:10:46 +0900 Subject: [PATCH 05/15] [ZEPPELIN-6683] Add msgId to notebook WebSocket replies --- .../zeppelin/socket/NotebookServer.java | 24 +++++++++++++++---- 1 file changed, 19 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 4d31a06558c..c9b0808b2cc 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 @@ -696,7 +696,10 @@ public void getInterpreterBindings(NotebookSocket conn, setting.getInterpreterInfos(), true)); } } - conn.send(serializeMessage(new Message(OP.INTERPRETER_BINDINGS).put("interpreterBindings", settingList))); + conn.send(serializeMessage( + new Message(OP.INTERPRETER_BINDINGS) + .withMsgId(fromMessage.msgId) + .put("interpreterBindings", settingList))); return null; }); } @@ -734,7 +737,9 @@ public void saveInterpreterBindings(NotebookSocket conn, ServiceContext context, }); if (permitted) { conn.send(serializeMessage( - new Message(OP.INTERPRETER_BINDINGS).put("interpreterBindings", settingList))); + new Message(OP.INTERPRETER_BINDINGS) + .withMsgId(fromMessage.msgId) + .put("interpreterBindings", settingList))); } } @@ -1669,7 +1674,10 @@ public void onSuccess(Revision revision, ServiceContext context) throws IOExcept List revisions = getNotebook().processNote(noteId, note -> getNotebook().listRevisionHistory(noteId, note.getPath(), context.getAutheInfo())); - conn.send(serializeMessage(new Message(OP.LIST_REVISION_HISTORY).put("revisionList", revisions))); + conn.send(serializeMessage( + new Message(OP.LIST_REVISION_HISTORY) + .withMsgId(fromMessage.msgId) + .put("revisionList", revisions))); } else { conn.send(serializeMessage( new Message(OP.ERROR_INFO).put("info", @@ -1689,7 +1697,10 @@ private void listRevisionHistory(NotebookSocket conn, @Override public void onSuccess(List revisions, ServiceContext context) throws IOException { super.onSuccess(revisions, context); - conn.send(serializeMessage(new Message(OP.LIST_REVISION_HISTORY).put("revisionList", revisions))); + conn.send(serializeMessage( + new Message(OP.LIST_REVISION_HISTORY) + .withMsgId(fromMessage.msgId) + .put("revisionList", revisions))); } }); } @@ -1705,7 +1716,10 @@ private void setNoteRevision(NotebookSocket conn, public void onSuccess(Note note, ServiceContext context) throws IOException { super.onSuccess(note, context); Note reloadedNote = getNotebook().loadNoteFromRepo(noteId, context.getAutheInfo()); - conn.send(serializeMessage(new Message(OP.SET_NOTE_REVISION).put("status", true))); + conn.send(serializeMessage( + new Message(OP.SET_NOTE_REVISION) + .withMsgId(fromMessage.msgId) + .put("status", true))); broadcastNote(reloadedNote); } }); From 1c0a7f883d96bdec36c5f51d92bb30a48d67a735 Mon Sep 17 00:00:00 2001 From: gyowoo1113 Date: Thu, 17 Sep 2026 14:04:50 +0900 Subject: [PATCH 06/15] [ZEPPELIN-6683] Add WebSocket msgId preservation tests --- .../zeppelin/socket/NotebookServerTest.java | 115 ++++++++++++++++-- 1 file changed, 108 insertions(+), 7 deletions(-) diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java index d288851fbff..c4ba63e1b02 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java @@ -1061,21 +1061,27 @@ void getInterpreterBindingsRequiresReaderPermission() throws IOException { try { setNotePermissions(noteId, "binding-owner", "binding-owner"); NotebookSocket socket = createWebSocket(); - Message message = new Message(OP.GET_INTERPRETER_BINDINGS).put("noteId", noteId); + + String msgId = "get-interpreter-bindings-msg"; + Message message = new Message(OP.GET_INTERPRETER_BINDINGS) + .withMsgId(msgId) + .put("noteId", noteId); notebookServer.getInterpreterBindings(socket, serviceContext("binding-attacker"), message); ArgumentCaptor response = ArgumentCaptor.forClass(String.class); verify(socket).send(response.capture()); - assertEquals(OP.AUTH_INFO, notebookServer.deserializeMessage(response.getValue()).op); + Message responseMessage = notebookServer.deserializeMessage(response.getValue()); + assertEquals(OP.AUTH_INFO, responseMessage.op); reset(socket); setNotePermissions(noteId, "binding-owner", "binding-reader"); notebookServer.getInterpreterBindings(socket, serviceContext("binding-reader"), message); verify(socket).send(response.capture()); - assertEquals(OP.INTERPRETER_BINDINGS, - notebookServer.deserializeMessage(response.getValue()).op); + responseMessage = notebookServer.deserializeMessage(response.getValue()); + assertEquals(OP.INTERPRETER_BINDINGS, responseMessage.op); + assertEquals(msgId, responseMessage.msgId); } finally { notebook.removeNote(noteId, owner); } @@ -1089,7 +1095,9 @@ void saveInterpreterBindingsRequiresWriterPermission() throws IOException { setNotePermissions(noteId, "binding-owner", "binding-reader"); String initialGroup = notebook.processNote(noteId, Note::getDefaultInterpreterGroup); String replacementGroup = initialGroup.equals("md") ? "spark" : "md"; + String msgId = "save-interpreter-bindings-msg"; Message message = new Message(OP.SAVE_INTERPRETER_BINDINGS) + .withMsgId(msgId) .put("noteId", noteId) .put("selectedSettingIds", Arrays.asList(replacementGroup)); NotebookSocket socket = createWebSocket(); @@ -1100,7 +1108,8 @@ void saveInterpreterBindingsRequiresWriterPermission() throws IOException { notebook.processNote(noteId, Note::getDefaultInterpreterGroup)); ArgumentCaptor response = ArgumentCaptor.forClass(String.class); verify(socket).send(response.capture()); - assertEquals(OP.AUTH_INFO, notebookServer.deserializeMessage(response.getValue()).op); + Message responseMessage = notebookServer.deserializeMessage(response.getValue()); + assertEquals(OP.AUTH_INFO, responseMessage.op); reset(socket); authorizationService.setWriters(noteId, @@ -1110,8 +1119,9 @@ void saveInterpreterBindingsRequiresWriterPermission() throws IOException { assertEquals(replacementGroup, notebook.processNote(noteId, Note::getDefaultInterpreterGroup)); verify(socket).send(response.capture()); - assertEquals(OP.INTERPRETER_BINDINGS, - notebookServer.deserializeMessage(response.getValue()).op); + responseMessage = notebookServer.deserializeMessage(response.getValue()); + assertEquals(OP.INTERPRETER_BINDINGS, responseMessage.op); + assertEquals(msgId, responseMessage.msgId); } finally { notebook.removeNote(noteId, owner); } @@ -1170,6 +1180,97 @@ void testNoteRevision() throws IOException { } } + @Test + void listRevisionHistoryPreservesMsgId() throws IOException { + String noteId = notebook.createNote("revision-msgid", anonymous); + + try { + NotebookSocket socket = createWebSocket(); + String msgId = "list-revision-history-msg"; + + Message request = new Message(OP.LIST_REVISION_HISTORY) + .withMsgId(msgId) + .put("noteId", noteId); + + notebookServer.onMessage(socket, request.toJson()); + + ArgumentCaptor response = ArgumentCaptor.forClass(String.class); + verify(socket).send(response.capture()); + + Message responseMessage = + notebookServer.deserializeMessage(response.getValue()); + + assertEquals(OP.LIST_REVISION_HISTORY, responseMessage.op); + assertEquals(msgId, responseMessage.msgId); + } finally { + notebook.removeNote(noteId, anonymous); + } + } + + @Test + void checkpointNotePreservesMsgId() throws IOException { + String noteId = notebook.createNote("checkpoint-msgid", anonymous); + + try { + NotebookSocket socket = createWebSocket(); + String msgId = "checkpoint-note-msg"; + + Message request = new Message(OP.CHECKPOINT_NOTE) + .withMsgId(msgId) + .put("noteId", noteId) + .put("commitMessage", "checkpoint"); + + notebookServer.onMessage(socket, request.toJson()); + + ArgumentCaptor response = ArgumentCaptor.forClass(String.class); + verify(socket).send(response.capture()); + + Message responseMessage = + notebookServer.deserializeMessage(response.getValue()); + + assertEquals(OP.LIST_REVISION_HISTORY, responseMessage.op); + assertEquals(msgId, responseMessage.msgId); + } finally { + notebook.removeNote(noteId, anonymous); + } + } + + @Test + void setNoteRevisionPreservesMsgId() throws IOException { + String noteId = notebook.createNote("set-revision-msgid", anonymous); + + try { + NotebookRepoWithVersionControl.Revision revision = + notebook.processNote(noteId, + note -> notebook.checkpointNote( + note.getId(), + note.getPath(), + "first commit", + anonymous)); + + NotebookSocket socket = createWebSocket(); + String msgId = "set-note-revision-msg"; + + Message request = new Message(OP.SET_NOTE_REVISION) + .withMsgId(msgId) + .put("noteId", noteId) + .put("revisionId", revision.id); + + notebookServer.onMessage(socket, request.toJson()); + + ArgumentCaptor response = ArgumentCaptor.forClass(String.class); + verify(socket).send(response.capture()); + + Message responseMessage = + notebookServer.deserializeMessage(response.getValue()); + + assertEquals(OP.SET_NOTE_REVISION, responseMessage.op); + assertEquals(msgId, responseMessage.msgId); + } finally { + notebook.removeNote(noteId, anonymous); + } + } + private NotebookSocket createWebSocket() { NotebookSocket sock = mock(NotebookSocket.class); return sock; From 1f02827843daf7200718f8d5cb427dd851a92d24 Mon Sep 17 00:00:00 2001 From: gyowoo1113 Date: Thu, 17 Sep 2026 16:52:00 +0900 Subject: [PATCH 07/15] [ZEPPELIN-6683] Add client-side stale reply tests --- .../projects/zeppelin-sdk/src/message.spec.ts | 18 +++ .../message-listener/message-listener.spec.ts | 38 ++++- .../notebook/notebook.component.spec.ts | 152 ++++++++++++++++++ .../src/app/services/message.service.spec.ts | 94 +++++++++++ 4 files changed, 299 insertions(+), 3 deletions(-) create mode 100644 zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts create mode 100644 zeppelin-web-angular/src/app/services/message.service.spec.ts diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.spec.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.spec.ts index 0bb02529cc7..37f8d44c2bf 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.spec.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.spec.ts @@ -143,4 +143,22 @@ describe('Message.receive', () => { expect(listener).toHaveBeenCalledWith(data); }); + + it('passes the full message envelope with msgId', () => { + const message = new Message(); + const listener = vi.fn(); + const data = {}; + + message.receiveEnvelope(OP.NOTE).subscribe(listener); + + const envelope = asReceivedMessage({ + op: OP.NOTE, + msgId: 'note-request-1', + data + }); + + message.shortCircuit(envelope); + + expect(listener).toHaveBeenCalledWith(envelope); + }); }); diff --git a/zeppelin-web-angular/src/app/core/message-listener/message-listener.spec.ts b/zeppelin-web-angular/src/app/core/message-listener/message-listener.spec.ts index e12767691d6..7fd1778ae6b 100644 --- a/zeppelin-web-angular/src/app/core/message-listener/message-listener.spec.ts +++ b/zeppelin-web-angular/src/app/core/message-listener/message-listener.spec.ts @@ -13,9 +13,8 @@ import { Subject } from 'rxjs'; import { afterEach, describe, expect, it, vi } from 'vitest'; -import { Message, OP, MessageReceiveDataTypeMap } from '@zeppelin/sdk'; - -import { MessageListener, MessageListenersManager } from './message-listener'; +import { Message, OP, MessageReceiveDataTypeMap, type WebSocketMessage } from '@zeppelin/sdk'; +import { MessageEnvelopeListener, MessageListener, MessageListenersManager } from './message-listener'; afterEach(() => { vi.restoreAllMocks(); @@ -88,3 +87,36 @@ describe('MessageListener', () => { expect(component.receivedData).toBe(data); }); }); + +describe('MessageEnvelopeListener', () => { + it('passes the received message envelope with msgId to the handler', () => { + const received$ = new Subject>(); + const messageService = { + receiveEnvelope: vi.fn(() => received$.asObservable()) + } as unknown as Message; + + class TestComponent extends MessageListenersManager { + receivedMessage?: WebSocketMessage; + + handleNote(message: WebSocketMessage): void { + this.receivedMessage = message; + } + } + + const descriptor = Object.getOwnPropertyDescriptor(TestComponent.prototype, 'handleNote')!; + + MessageEnvelopeListener(OP.NOTE)(TestComponent.prototype, 'handleNote', descriptor); + + const component = new TestComponent(messageService); + const envelope: WebSocketMessage = { + op: OP.NOTE, + msgId: 'note-request-1', + data: {} as MessageReceiveDataTypeMap[OP.NOTE] + }; + + received$.next(envelope); + + expect(component.receivedMessage).toBe(envelope); + expect(messageService.receiveEnvelope).toHaveBeenCalledWith(OP.NOTE); + }); +}); diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts new file mode 100644 index 00000000000..3a0e8f568e8 --- /dev/null +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts @@ -0,0 +1,152 @@ +import { NEVER } from 'rxjs'; +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import type { ChangeDetectorRef } from '@angular/core'; +import type { Title } from '@angular/platform-browser'; +import type { ActivatedRoute, Router } from '@angular/router'; +import { OP, type MessageReceiveDataTypeMap, type WebSocketMessage } from '@zeppelin/sdk'; + +import type { + MessageService, + NgZService, + NoteStatusService, + NoteVarShareService, + ReactFeatureService, + SecurityService, + ThemeService, + TicketService +} from '@zeppelin/services'; + +vi.mock('./paragraph/paragraph.component', () => ({ + NotebookParagraphComponent: class NotebookParagraphComponent {} +})); + +import { NotebookComponent } from './notebook.component'; + +const createComponent = () => { + const messageService = { + consumePendingNoteRequest: vi.fn(), + receive: vi.fn(() => NEVER), + receiveEnvelope: vi.fn(() => NEVER) + } as unknown as MessageService; + + const router = { + navigate: vi.fn(() => Promise.resolve(true)) + } as unknown as Router; + + const component = new NotebookComponent( + messageService, + {} as NgZService, + { + snapshot: { + params: { + noteId: 'note-b' + } + } + } as unknown as ActivatedRoute, + { + markForCheck: vi.fn() + } as unknown as ChangeDetectorRef, + {} as NoteStatusService, + {} as NoteVarShareService, + {} as TicketService, + {} as SecurityService, + router, + {} as Title, + {} as ThemeService, + {} as ReactFeatureService + ); + + return { + component, + messageService, + router + }; +}; + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe('NotebookComponent', () => { + it('ignores stale interpreter bindings replies', () => { + const { component, messageService } = createComponent(); + + vi.mocked(messageService.consumePendingNoteRequest).mockReturnValue(false); + + const originalBindings = [{ id: 'existing', selected: true }]; + component.interpreterBindings = originalBindings as typeof component.interpreterBindings; + + const message: WebSocketMessage = { + op: OP.INTERPRETER_BINDINGS, + msgId: 'msg-a', + data: { + interpreterBindings: [{ id: 'stale', selected: false }] + } + }; + + component.loadInterpreterBindings(message); + + expect(messageService.consumePendingNoteRequest).toHaveBeenCalledWith('msg-a', 'note-b'); + expect(component.interpreterBindings).toBe(originalBindings); + }); + + it('ignores stale revision history replies', () => { + const { component, messageService } = createComponent(); + + vi.mocked(messageService.consumePendingNoteRequest).mockReturnValue(false); + + const originalRevisions = [{ id: 'Head', message: 'Head' }]; + component.noteRevisions = originalRevisions; + component.currentRevision = 'Head'; + + const message: WebSocketMessage = { + op: OP.LIST_REVISION_HISTORY, + msgId: 'msg-a', + data: { + revisionList: [{ id: 'rev-1', message: 'stale revision' }] + } + }; + component.listRevisionHistory(message); + + expect(messageService.consumePendingNoteRequest).toHaveBeenCalledWith('msg-a', 'note-b'); + expect(component.noteRevisions).toBe(originalRevisions); + expect(component.currentRevision).toBe('Head'); + }); + + it('ignores stale set note revision replies', () => { + const { component, messageService, router } = createComponent(); + + vi.mocked(messageService.consumePendingNoteRequest).mockReturnValue(false); + + const message: WebSocketMessage = { + op: OP.SET_NOTE_REVISION, + msgId: 'msg-a', + data: undefined + }; + + component.setNoteRevision(message); + + expect(messageService.consumePendingNoteRequest).toHaveBeenCalledWith('msg-a', 'note-b'); + expect(router.navigate).not.toHaveBeenCalled(); + }); + + it('applies interpreter bindings for the active note', () => { + const { component, messageService } = createComponent(); + + vi.mocked(messageService.consumePendingNoteRequest).mockReturnValue(true); + + const message: WebSocketMessage = { + op: OP.INTERPRETER_BINDINGS, + msgId: 'msg-b', + data: { + interpreterBindings: [{ id: 'current', selected: true }] + } + }; + + component.loadInterpreterBindings(message); + + expect(messageService.consumePendingNoteRequest).toHaveBeenCalledWith('msg-b', 'note-b'); + expect(component.interpreterBindings).toEqual([{ id: 'current', selected: true }]); + }); +}); diff --git a/zeppelin-web-angular/src/app/services/message.service.spec.ts b/zeppelin-web-angular/src/app/services/message.service.spec.ts new file mode 100644 index 00000000000..65865f4aae0 --- /dev/null +++ b/zeppelin-web-angular/src/app/services/message.service.spec.ts @@ -0,0 +1,94 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import type { MessageInterceptor } from '@zeppelin/interfaces'; +import { MessageService } from './message.service'; +import { BaseUrlService } from './base-url.service'; +import { TicketService } from './ticket.service'; + +const createService = (): MessageService => + new MessageService( + { getWebsocketUrl: vi.fn() } as unknown as BaseUrlService, + { originTicket: {} } as unknown as TicketService, + undefined as unknown as MessageInterceptor + ); + +const getPendingNoteRequests = (service: MessageService): Map => + ( + service as unknown as { + pendingNoteRequests: Map; + } + ).pendingNoteRequests; + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe('MessageService', () => { + it('rejects a missing pending request id', () => { + const service = createService(); + + expect(service.consumePendingNoteRequest(undefined, 'note-a')).toBe(false); + }); + + it('rejects an unknown pending request id', () => { + const service = createService(); + + expect(service.consumePendingNoteRequest('unknown-msg-id', 'note-a')).toBe(false); + }); + + it('accepts a pending request for the active note', () => { + const service = createService(); + + const pendingNoteRequests = getPendingNoteRequests(service); + + pendingNoteRequests.set('msg-1', 'note-a'); + + expect(service.consumePendingNoteRequest('msg-1', 'note-a')).toBe(true); + }); + + it('rejects a pending request for a different active note', () => { + const service = createService(); + + const pendingNoteRequests = getPendingNoteRequests(service); + + pendingNoteRequests.set('msg-1', 'note-a'); + + expect(service.consumePendingNoteRequest('msg-1', 'note-b')).toBe(false); + expect(service.consumePendingNoteRequest('msg-1', 'note-a')).toBe(false); + }); + + it('distinguishes two outstanding requests by message id', () => { + const service = createService(); + + const pendingNoteRequests = getPendingNoteRequests(service); + + pendingNoteRequests.set('msg-1', 'note-a'); + pendingNoteRequests.set('msg-2', 'note-b'); + + expect(service.consumePendingNoteRequest('msg-1', 'note-a')).toBe(true); + expect(service.consumePendingNoteRequest('msg-2', 'note-b')).toBe(true); + }); + + it('records the note context when sending an interpreter bindings request', () => { + const service = createService(); + + const ws = { + next: vi.fn() + }; + + ( + service as unknown as { + ws: { next: (message: unknown) => void }; + } + ).ws = ws; + + service.getInterpreterBindings('note-a'); + + const sentMessage = ws.next.mock.calls[0][0] as { + msgId?: string; + }; + + expect(sentMessage.msgId).toBeDefined(); + expect(service.consumePendingNoteRequest(sentMessage.msgId, 'note-a')).toBe(true); + }); +}); From 4fb1918e2af851214b81e14554d33ebef6de97a0 Mon Sep 17 00:00:00 2001 From: gyowoo1113 Date: Thu, 17 Sep 2026 17:12:16 +0900 Subject: [PATCH 08/15] [ZEPPELIN-6683] add license header --- .../workspace/notebook/notebook.component.spec.ts | 12 ++++++++++++ .../src/app/services/message.service.spec.ts | 12 ++++++++++++ 2 files changed, 24 insertions(+) diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts index 3a0e8f568e8..ea83b427fd3 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts @@ -1,3 +1,15 @@ +/* + * 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 { NEVER } from 'rxjs'; import { afterEach, describe, expect, it, vi } from 'vitest'; diff --git a/zeppelin-web-angular/src/app/services/message.service.spec.ts b/zeppelin-web-angular/src/app/services/message.service.spec.ts index 65865f4aae0..e41c3146d73 100644 --- a/zeppelin-web-angular/src/app/services/message.service.spec.ts +++ b/zeppelin-web-angular/src/app/services/message.service.spec.ts @@ -1,3 +1,15 @@ +/* + * 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 { afterEach, describe, expect, it, vi } from 'vitest'; import type { MessageInterceptor } from '@zeppelin/interfaces'; From 7724dcd3ea7c6eefde65c18ec52990f0a7988ca7 Mon Sep 17 00:00:00 2001 From: gyowoo1113 Date: Tue, 22 Sep 2026 10:07:58 +0900 Subject: [PATCH 09/15] Revert "[ZEPPELIN-6683] Add WebSocket msgId preservation tests" --- .../zeppelin/socket/NotebookServerTest.java | 115 ++---------------- 1 file changed, 7 insertions(+), 108 deletions(-) diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java index c4ba63e1b02..d288851fbff 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java @@ -1061,27 +1061,21 @@ void getInterpreterBindingsRequiresReaderPermission() throws IOException { try { setNotePermissions(noteId, "binding-owner", "binding-owner"); NotebookSocket socket = createWebSocket(); - - String msgId = "get-interpreter-bindings-msg"; - Message message = new Message(OP.GET_INTERPRETER_BINDINGS) - .withMsgId(msgId) - .put("noteId", noteId); + Message message = new Message(OP.GET_INTERPRETER_BINDINGS).put("noteId", noteId); notebookServer.getInterpreterBindings(socket, serviceContext("binding-attacker"), message); ArgumentCaptor response = ArgumentCaptor.forClass(String.class); verify(socket).send(response.capture()); - Message responseMessage = notebookServer.deserializeMessage(response.getValue()); - assertEquals(OP.AUTH_INFO, responseMessage.op); + assertEquals(OP.AUTH_INFO, notebookServer.deserializeMessage(response.getValue()).op); reset(socket); setNotePermissions(noteId, "binding-owner", "binding-reader"); notebookServer.getInterpreterBindings(socket, serviceContext("binding-reader"), message); verify(socket).send(response.capture()); - responseMessage = notebookServer.deserializeMessage(response.getValue()); - assertEquals(OP.INTERPRETER_BINDINGS, responseMessage.op); - assertEquals(msgId, responseMessage.msgId); + assertEquals(OP.INTERPRETER_BINDINGS, + notebookServer.deserializeMessage(response.getValue()).op); } finally { notebook.removeNote(noteId, owner); } @@ -1095,9 +1089,7 @@ void saveInterpreterBindingsRequiresWriterPermission() throws IOException { setNotePermissions(noteId, "binding-owner", "binding-reader"); String initialGroup = notebook.processNote(noteId, Note::getDefaultInterpreterGroup); String replacementGroup = initialGroup.equals("md") ? "spark" : "md"; - String msgId = "save-interpreter-bindings-msg"; Message message = new Message(OP.SAVE_INTERPRETER_BINDINGS) - .withMsgId(msgId) .put("noteId", noteId) .put("selectedSettingIds", Arrays.asList(replacementGroup)); NotebookSocket socket = createWebSocket(); @@ -1108,8 +1100,7 @@ void saveInterpreterBindingsRequiresWriterPermission() throws IOException { notebook.processNote(noteId, Note::getDefaultInterpreterGroup)); ArgumentCaptor response = ArgumentCaptor.forClass(String.class); verify(socket).send(response.capture()); - Message responseMessage = notebookServer.deserializeMessage(response.getValue()); - assertEquals(OP.AUTH_INFO, responseMessage.op); + assertEquals(OP.AUTH_INFO, notebookServer.deserializeMessage(response.getValue()).op); reset(socket); authorizationService.setWriters(noteId, @@ -1119,9 +1110,8 @@ void saveInterpreterBindingsRequiresWriterPermission() throws IOException { assertEquals(replacementGroup, notebook.processNote(noteId, Note::getDefaultInterpreterGroup)); verify(socket).send(response.capture()); - responseMessage = notebookServer.deserializeMessage(response.getValue()); - assertEquals(OP.INTERPRETER_BINDINGS, responseMessage.op); - assertEquals(msgId, responseMessage.msgId); + assertEquals(OP.INTERPRETER_BINDINGS, + notebookServer.deserializeMessage(response.getValue()).op); } finally { notebook.removeNote(noteId, owner); } @@ -1180,97 +1170,6 @@ void testNoteRevision() throws IOException { } } - @Test - void listRevisionHistoryPreservesMsgId() throws IOException { - String noteId = notebook.createNote("revision-msgid", anonymous); - - try { - NotebookSocket socket = createWebSocket(); - String msgId = "list-revision-history-msg"; - - Message request = new Message(OP.LIST_REVISION_HISTORY) - .withMsgId(msgId) - .put("noteId", noteId); - - notebookServer.onMessage(socket, request.toJson()); - - ArgumentCaptor response = ArgumentCaptor.forClass(String.class); - verify(socket).send(response.capture()); - - Message responseMessage = - notebookServer.deserializeMessage(response.getValue()); - - assertEquals(OP.LIST_REVISION_HISTORY, responseMessage.op); - assertEquals(msgId, responseMessage.msgId); - } finally { - notebook.removeNote(noteId, anonymous); - } - } - - @Test - void checkpointNotePreservesMsgId() throws IOException { - String noteId = notebook.createNote("checkpoint-msgid", anonymous); - - try { - NotebookSocket socket = createWebSocket(); - String msgId = "checkpoint-note-msg"; - - Message request = new Message(OP.CHECKPOINT_NOTE) - .withMsgId(msgId) - .put("noteId", noteId) - .put("commitMessage", "checkpoint"); - - notebookServer.onMessage(socket, request.toJson()); - - ArgumentCaptor response = ArgumentCaptor.forClass(String.class); - verify(socket).send(response.capture()); - - Message responseMessage = - notebookServer.deserializeMessage(response.getValue()); - - assertEquals(OP.LIST_REVISION_HISTORY, responseMessage.op); - assertEquals(msgId, responseMessage.msgId); - } finally { - notebook.removeNote(noteId, anonymous); - } - } - - @Test - void setNoteRevisionPreservesMsgId() throws IOException { - String noteId = notebook.createNote("set-revision-msgid", anonymous); - - try { - NotebookRepoWithVersionControl.Revision revision = - notebook.processNote(noteId, - note -> notebook.checkpointNote( - note.getId(), - note.getPath(), - "first commit", - anonymous)); - - NotebookSocket socket = createWebSocket(); - String msgId = "set-note-revision-msg"; - - Message request = new Message(OP.SET_NOTE_REVISION) - .withMsgId(msgId) - .put("noteId", noteId) - .put("revisionId", revision.id); - - notebookServer.onMessage(socket, request.toJson()); - - ArgumentCaptor response = ArgumentCaptor.forClass(String.class); - verify(socket).send(response.capture()); - - Message responseMessage = - notebookServer.deserializeMessage(response.getValue()); - - assertEquals(OP.SET_NOTE_REVISION, responseMessage.op); - assertEquals(msgId, responseMessage.msgId); - } finally { - notebook.removeNote(noteId, anonymous); - } - } - private NotebookSocket createWebSocket() { NotebookSocket sock = mock(NotebookSocket.class); return sock; From 2ce775303d194fa4758f2c8394348eb6fdb8c7e7 Mon Sep 17 00:00:00 2001 From: gyowoo1113 Date: Tue, 22 Sep 2026 10:08:06 +0900 Subject: [PATCH 10/15] Revert "[ZEPPELIN-6683] Add msgId to notebook WebSocket replies" This reverts commit db48085fe0421d7e8378faa551e896791e442efb. --- .../zeppelin/socket/NotebookServer.java | 24 ++++--------------- 1 file changed, 5 insertions(+), 19 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 c9b0808b2cc..4d31a06558c 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 @@ -696,10 +696,7 @@ public void getInterpreterBindings(NotebookSocket conn, setting.getInterpreterInfos(), true)); } } - conn.send(serializeMessage( - new Message(OP.INTERPRETER_BINDINGS) - .withMsgId(fromMessage.msgId) - .put("interpreterBindings", settingList))); + conn.send(serializeMessage(new Message(OP.INTERPRETER_BINDINGS).put("interpreterBindings", settingList))); return null; }); } @@ -737,9 +734,7 @@ public void saveInterpreterBindings(NotebookSocket conn, ServiceContext context, }); if (permitted) { conn.send(serializeMessage( - new Message(OP.INTERPRETER_BINDINGS) - .withMsgId(fromMessage.msgId) - .put("interpreterBindings", settingList))); + new Message(OP.INTERPRETER_BINDINGS).put("interpreterBindings", settingList))); } } @@ -1674,10 +1669,7 @@ public void onSuccess(Revision revision, ServiceContext context) throws IOExcept List revisions = getNotebook().processNote(noteId, note -> getNotebook().listRevisionHistory(noteId, note.getPath(), context.getAutheInfo())); - conn.send(serializeMessage( - new Message(OP.LIST_REVISION_HISTORY) - .withMsgId(fromMessage.msgId) - .put("revisionList", revisions))); + conn.send(serializeMessage(new Message(OP.LIST_REVISION_HISTORY).put("revisionList", revisions))); } else { conn.send(serializeMessage( new Message(OP.ERROR_INFO).put("info", @@ -1697,10 +1689,7 @@ private void listRevisionHistory(NotebookSocket conn, @Override public void onSuccess(List revisions, ServiceContext context) throws IOException { super.onSuccess(revisions, context); - conn.send(serializeMessage( - new Message(OP.LIST_REVISION_HISTORY) - .withMsgId(fromMessage.msgId) - .put("revisionList", revisions))); + conn.send(serializeMessage(new Message(OP.LIST_REVISION_HISTORY).put("revisionList", revisions))); } }); } @@ -1716,10 +1705,7 @@ private void setNoteRevision(NotebookSocket conn, public void onSuccess(Note note, ServiceContext context) throws IOException { super.onSuccess(note, context); Note reloadedNote = getNotebook().loadNoteFromRepo(noteId, context.getAutheInfo()); - conn.send(serializeMessage( - new Message(OP.SET_NOTE_REVISION) - .withMsgId(fromMessage.msgId) - .put("status", true))); + conn.send(serializeMessage(new Message(OP.SET_NOTE_REVISION).put("status", true))); broadcastNote(reloadedNote); } }); From c5ffaca3ee0dfbbc3e2c4f68ea125ea235a61381 Mon Sep 17 00:00:00 2001 From: gyowoo1113 Date: Tue, 22 Sep 2026 14:32:18 +0900 Subject: [PATCH 11/15] [ZEPPELIN-6683] Remove pending WebSocket request correlation --- .../notebook/notebook.component.spec.ts | 164 ------------------ .../workspace/notebook/notebook.component.ts | 19 +- .../src/app/services/message.service.spec.ts | 106 ----------- .../src/app/services/message.service.ts | 56 ++---- 4 files changed, 15 insertions(+), 330 deletions(-) delete mode 100644 zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts delete mode 100644 zeppelin-web-angular/src/app/services/message.service.spec.ts diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts deleted file mode 100644 index ea83b427fd3..00000000000 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts +++ /dev/null @@ -1,164 +0,0 @@ -/* - * 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 { NEVER } from 'rxjs'; -import { afterEach, describe, expect, it, vi } from 'vitest'; - -import type { ChangeDetectorRef } from '@angular/core'; -import type { Title } from '@angular/platform-browser'; -import type { ActivatedRoute, Router } from '@angular/router'; -import { OP, type MessageReceiveDataTypeMap, type WebSocketMessage } from '@zeppelin/sdk'; - -import type { - MessageService, - NgZService, - NoteStatusService, - NoteVarShareService, - ReactFeatureService, - SecurityService, - ThemeService, - TicketService -} from '@zeppelin/services'; - -vi.mock('./paragraph/paragraph.component', () => ({ - NotebookParagraphComponent: class NotebookParagraphComponent {} -})); - -import { NotebookComponent } from './notebook.component'; - -const createComponent = () => { - const messageService = { - consumePendingNoteRequest: vi.fn(), - receive: vi.fn(() => NEVER), - receiveEnvelope: vi.fn(() => NEVER) - } as unknown as MessageService; - - const router = { - navigate: vi.fn(() => Promise.resolve(true)) - } as unknown as Router; - - const component = new NotebookComponent( - messageService, - {} as NgZService, - { - snapshot: { - params: { - noteId: 'note-b' - } - } - } as unknown as ActivatedRoute, - { - markForCheck: vi.fn() - } as unknown as ChangeDetectorRef, - {} as NoteStatusService, - {} as NoteVarShareService, - {} as TicketService, - {} as SecurityService, - router, - {} as Title, - {} as ThemeService, - {} as ReactFeatureService - ); - - return { - component, - messageService, - router - }; -}; - -afterEach(() => { - vi.restoreAllMocks(); -}); - -describe('NotebookComponent', () => { - it('ignores stale interpreter bindings replies', () => { - const { component, messageService } = createComponent(); - - vi.mocked(messageService.consumePendingNoteRequest).mockReturnValue(false); - - const originalBindings = [{ id: 'existing', selected: true }]; - component.interpreterBindings = originalBindings as typeof component.interpreterBindings; - - const message: WebSocketMessage = { - op: OP.INTERPRETER_BINDINGS, - msgId: 'msg-a', - data: { - interpreterBindings: [{ id: 'stale', selected: false }] - } - }; - - component.loadInterpreterBindings(message); - - expect(messageService.consumePendingNoteRequest).toHaveBeenCalledWith('msg-a', 'note-b'); - expect(component.interpreterBindings).toBe(originalBindings); - }); - - it('ignores stale revision history replies', () => { - const { component, messageService } = createComponent(); - - vi.mocked(messageService.consumePendingNoteRequest).mockReturnValue(false); - - const originalRevisions = [{ id: 'Head', message: 'Head' }]; - component.noteRevisions = originalRevisions; - component.currentRevision = 'Head'; - - const message: WebSocketMessage = { - op: OP.LIST_REVISION_HISTORY, - msgId: 'msg-a', - data: { - revisionList: [{ id: 'rev-1', message: 'stale revision' }] - } - }; - component.listRevisionHistory(message); - - expect(messageService.consumePendingNoteRequest).toHaveBeenCalledWith('msg-a', 'note-b'); - expect(component.noteRevisions).toBe(originalRevisions); - expect(component.currentRevision).toBe('Head'); - }); - - it('ignores stale set note revision replies', () => { - const { component, messageService, router } = createComponent(); - - vi.mocked(messageService.consumePendingNoteRequest).mockReturnValue(false); - - const message: WebSocketMessage = { - op: OP.SET_NOTE_REVISION, - msgId: 'msg-a', - data: undefined - }; - - component.setNoteRevision(message); - - expect(messageService.consumePendingNoteRequest).toHaveBeenCalledWith('msg-a', 'note-b'); - expect(router.navigate).not.toHaveBeenCalled(); - }); - - it('applies interpreter bindings for the active note', () => { - const { component, messageService } = createComponent(); - - vi.mocked(messageService.consumePendingNoteRequest).mockReturnValue(true); - - const message: WebSocketMessage = { - op: OP.INTERPRETER_BINDINGS, - msgId: 'msg-b', - data: { - interpreterBindings: [{ id: 'current', selected: true }] - } - }; - - component.loadInterpreterBindings(message); - - expect(messageService.consumePendingNoteRequest).toHaveBeenCalledWith('msg-b', 'note-b'); - expect(component.interpreterBindings).toEqual([{ id: 'current', selected: true }]); - }); -}); diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts index 2af9fca28e2..f0398cbae8e 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts @@ -119,12 +119,6 @@ export class NotebookComponent extends MessageListenersManager implements OnInit @MessageEnvelopeListener(OP.INTERPRETER_BINDINGS) loadInterpreterBindings(message: WebSocketMessage) { - const { noteId } = this.activatedRoute.snapshot.params; - - if (!this.messageService.consumePendingNoteRequest(message.msgId, noteId)) { - return; - } - const data = message.data; if (data === undefined) { return; @@ -210,13 +204,8 @@ export class NotebookComponent extends MessageListenersManager implements OnInit } @MessageEnvelopeListener(OP.SET_NOTE_REVISION) - setNoteRevision(message: WebSocketMessage) { + setNoteRevision(_message: WebSocketMessage) { const { noteId } = this.activatedRoute.snapshot.params; - - if (!this.messageService.consumePendingNoteRequest(message.msgId, noteId)) { - return; - } - this.router.navigate(['/notebook', noteId]).then(); } @@ -274,12 +263,6 @@ export class NotebookComponent extends MessageListenersManager implements OnInit @MessageEnvelopeListener(OP.LIST_REVISION_HISTORY) listRevisionHistory(message: WebSocketMessage) { - const { noteId } = this.activatedRoute.snapshot.params; - - if (!this.messageService.consumePendingNoteRequest(message.msgId, noteId)) { - return; - } - const data = message.data; if (data === undefined) { return; diff --git a/zeppelin-web-angular/src/app/services/message.service.spec.ts b/zeppelin-web-angular/src/app/services/message.service.spec.ts deleted file mode 100644 index e41c3146d73..00000000000 --- a/zeppelin-web-angular/src/app/services/message.service.spec.ts +++ /dev/null @@ -1,106 +0,0 @@ -/* - * 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 { afterEach, describe, expect, it, vi } from 'vitest'; - -import type { MessageInterceptor } from '@zeppelin/interfaces'; -import { MessageService } from './message.service'; -import { BaseUrlService } from './base-url.service'; -import { TicketService } from './ticket.service'; - -const createService = (): MessageService => - new MessageService( - { getWebsocketUrl: vi.fn() } as unknown as BaseUrlService, - { originTicket: {} } as unknown as TicketService, - undefined as unknown as MessageInterceptor - ); - -const getPendingNoteRequests = (service: MessageService): Map => - ( - service as unknown as { - pendingNoteRequests: Map; - } - ).pendingNoteRequests; - -afterEach(() => { - vi.restoreAllMocks(); -}); - -describe('MessageService', () => { - it('rejects a missing pending request id', () => { - const service = createService(); - - expect(service.consumePendingNoteRequest(undefined, 'note-a')).toBe(false); - }); - - it('rejects an unknown pending request id', () => { - const service = createService(); - - expect(service.consumePendingNoteRequest('unknown-msg-id', 'note-a')).toBe(false); - }); - - it('accepts a pending request for the active note', () => { - const service = createService(); - - const pendingNoteRequests = getPendingNoteRequests(service); - - pendingNoteRequests.set('msg-1', 'note-a'); - - expect(service.consumePendingNoteRequest('msg-1', 'note-a')).toBe(true); - }); - - it('rejects a pending request for a different active note', () => { - const service = createService(); - - const pendingNoteRequests = getPendingNoteRequests(service); - - pendingNoteRequests.set('msg-1', 'note-a'); - - expect(service.consumePendingNoteRequest('msg-1', 'note-b')).toBe(false); - expect(service.consumePendingNoteRequest('msg-1', 'note-a')).toBe(false); - }); - - it('distinguishes two outstanding requests by message id', () => { - const service = createService(); - - const pendingNoteRequests = getPendingNoteRequests(service); - - pendingNoteRequests.set('msg-1', 'note-a'); - pendingNoteRequests.set('msg-2', 'note-b'); - - expect(service.consumePendingNoteRequest('msg-1', 'note-a')).toBe(true); - expect(service.consumePendingNoteRequest('msg-2', 'note-b')).toBe(true); - }); - - it('records the note context when sending an interpreter bindings request', () => { - const service = createService(); - - const ws = { - next: vi.fn() - }; - - ( - service as unknown as { - ws: { next: (message: unknown) => void }; - } - ).ws = ws; - - service.getInterpreterBindings('note-a'); - - const sentMessage = ws.next.mock.calls[0][0] as { - msgId?: string; - }; - - expect(sentMessage.msgId).toBeDefined(); - expect(service.consumePendingNoteRequest(sentMessage.msgId, 'note-a')).toBe(true); - }); -}); diff --git a/zeppelin-web-angular/src/app/services/message.service.ts b/zeppelin-web-angular/src/app/services/message.service.ts index 3ceac51ec84..050d78c44cc 100644 --- a/zeppelin-web-angular/src/app/services/message.service.ts +++ b/zeppelin-web-angular/src/app/services/message.service.ts @@ -41,7 +41,6 @@ import { TicketService } from './ticket.service'; }) export class MessageService extends Message implements OnDestroy { private readonly localAddFocusMsgIds = new Set(); - private readonly pendingNoteRequests = new Map(); constructor( private baseUrlService: BaseUrlService, @@ -94,26 +93,15 @@ export class MessageService extends Message implements OnDestroy { return this.localAddFocusMsgIds.delete(msgId); } - consumePendingNoteRequest(msgId: string | undefined, activeNoteId: string): boolean { - if (!msgId) { - return false; - } - - const requestedNoteId = this.pendingNoteRequests.get(msgId); - if (!requestedNoteId) { - return false; - } - - this.pendingNoteRequests.delete(msgId); - - return requestedNoteId === activeNoteId; - } - - private captureSentMessage( - sendMessage: () => void, - onSent: (message: WebSocketMessage) => void - ): void { - const subscription = super.sent().pipe(take(1)).subscribe(onSent); + private captureLocalAddFocusMsgId(sendMessage: () => void): void { + const subscription = super + .sent() + .pipe(take(1)) + .subscribe(message => { + if (message.msgId) { + this.localAddFocusMsgIds.add(message.msgId); + } + }); try { sendMessage(); } catch (error) { @@ -122,22 +110,6 @@ export class MessageService extends Message implements OnDestroy { } } - private captureLocalAddFocusMsgId(sendMessage: () => void): void { - this.captureSentMessage(sendMessage, message => { - if (message.msgId) { - this.localAddFocusMsgIds.add(message.msgId); - } - }); - } - - private capturePendingNoteRequest(noteId: string, sendMessage: () => void): void { - this.captureSentMessage(sendMessage, message => { - if (message.msgId) { - this.pendingNoteRequests.set(message.msgId, noteId); - } - }); - } - opened(): Observable { return super.opened(); } @@ -345,15 +317,15 @@ export class MessageService extends Message implements OnDestroy { } checkpointNote(noteId: string, commitMessage: string): void { - this.capturePendingNoteRequest(noteId, () => super.checkpointNote(noteId, commitMessage)); + super.checkpointNote(noteId, commitMessage); } setNoteRevision(noteId: string, revisionId: string): void { - this.capturePendingNoteRequest(noteId, () => super.setNoteRevision(noteId, revisionId)); + super.setNoteRevision(noteId, revisionId); } listRevisionHistory(noteId: string): void { - this.capturePendingNoteRequest(noteId, () => super.listRevisionHistory(noteId)); + super.listRevisionHistory(noteId); } noteRevision(noteId: string, revisionId: string): void { @@ -377,11 +349,11 @@ export class MessageService extends Message implements OnDestroy { } getInterpreterBindings(noteId: string): void { - this.capturePendingNoteRequest(noteId, () => super.getInterpreterBindings(noteId)); + super.getInterpreterBindings(noteId); } saveInterpreterBindings(noteId: string, selectedSettingIds: string[]): void { - this.capturePendingNoteRequest(noteId, () => super.saveInterpreterBindings(noteId, selectedSettingIds)); + super.saveInterpreterBindings(noteId, selectedSettingIds); } getInterpreterSettings(): void { From 8e5b926357461b5496fbcc9102365589947dca99 Mon Sep 17 00:00:00 2001 From: gyowoo1113 Date: Tue, 22 Sep 2026 16:01:29 +0900 Subject: [PATCH 12/15] Revert "[ZEPPELIN-6683] Expose WebSocket reply envelopes to notebook listeners" --- .../workspace/notebook/notebook.component.ts | 27 ++++++------------- 1 file changed, 8 insertions(+), 19 deletions(-) diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts index f0398cbae8e..3dd6e739607 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts @@ -28,7 +28,7 @@ import { distinctUntilChanged, distinctUntilKeyChanged, startWith, takeUntil } f import { NzResizeEvent } from 'ng-zorro-antd/resizable'; -import { MessageEnvelopeListener, MessageListener, MessageListenersManager } from '@zeppelin/core'; +import { MessageListener, MessageListenersManager } from '@zeppelin/core'; import { Permissions } from '@zeppelin/interfaces'; import { DynamicFormParams, @@ -36,8 +36,7 @@ import { MessageReceiveDataTypeMap, Note, OP, - RevisionListItem, - WebSocketMessage + RevisionListItem } from '@zeppelin/sdk'; import { MessageService, @@ -117,13 +116,8 @@ export class NotebookComponent extends MessageListenersManager implements OnInit } } - @MessageEnvelopeListener(OP.INTERPRETER_BINDINGS) - loadInterpreterBindings(message: WebSocketMessage) { - const data = message.data; - if (data === undefined) { - return; - } - + @MessageListener(OP.INTERPRETER_BINDINGS) + loadInterpreterBindings(data: MessageReceiveDataTypeMap[OP.INTERPRETER_BINDINGS]) { this.interpreterBindings = data.interpreterBindings; if (!this.interpreterBindings.some(item => item.selected)) { this.activatedExtension = 'interpreter'; @@ -203,8 +197,8 @@ export class NotebookComponent extends MessageListenersManager implements OnInit } } - @MessageEnvelopeListener(OP.SET_NOTE_REVISION) - setNoteRevision(_message: WebSocketMessage) { + @MessageListener(OP.SET_NOTE_REVISION) + setNoteRevision(_data: MessageReceiveDataTypeMap[OP.SET_NOTE_REVISION]) { const { noteId } = this.activatedRoute.snapshot.params; this.router.navigate(['/notebook', noteId]).then(); } @@ -261,13 +255,8 @@ export class NotebookComponent extends MessageListenersManager implements OnInit this.cdr.markForCheck(); } - @MessageEnvelopeListener(OP.LIST_REVISION_HISTORY) - listRevisionHistory(message: WebSocketMessage) { - const data = message.data; - if (data === undefined) { - return; - } - + @MessageListener(OP.LIST_REVISION_HISTORY) + listRevisionHistory(data: MessageReceiveDataTypeMap[OP.LIST_REVISION_HISTORY]) { this.noteRevisions = data.revisionList; if (this.noteRevisions) { if (this.noteRevisions.length === 0 || this.noteRevisions[0].id !== 'Head') { From c3f18ede9a1dd746709f92fbe6b31c3624e4d248 Mon Sep 17 00:00:00 2001 From: gyowoo1113 Date: Tue, 22 Sep 2026 16:21:32 +0900 Subject: [PATCH 13/15] [ZEPPELIN-6683] Guard notebook replies by noteId --- .../zeppelin/socket/NotebookServer.java | 20 ++++++++++++++----- .../message-interpreter.interface.ts | 1 + .../interfaces/message-notebook.interface.ts | 2 ++ .../workspace/notebook/notebook.component.ts | 15 +++++++++++++- 4 files changed, 32 insertions(+), 6 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 4d31a06558c..8a2219d3219 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 @@ -696,7 +696,9 @@ public void getInterpreterBindings(NotebookSocket conn, setting.getInterpreterInfos(), true)); } } - conn.send(serializeMessage(new Message(OP.INTERPRETER_BINDINGS).put("interpreterBindings", settingList))); + conn.send(serializeMessage(new Message(OP.INTERPRETER_BINDINGS) + .put("noteId", noteId) + .put("interpreterBindings", settingList))); return null; }); } @@ -734,7 +736,9 @@ public void saveInterpreterBindings(NotebookSocket conn, ServiceContext context, }); if (permitted) { conn.send(serializeMessage( - new Message(OP.INTERPRETER_BINDINGS).put("interpreterBindings", settingList))); + new Message(OP.INTERPRETER_BINDINGS) + .put("noteId", noteId) + .put("interpreterBindings", settingList))); } } @@ -1669,7 +1673,9 @@ public void onSuccess(Revision revision, ServiceContext context) throws IOExcept List revisions = getNotebook().processNote(noteId, note -> getNotebook().listRevisionHistory(noteId, note.getPath(), context.getAutheInfo())); - conn.send(serializeMessage(new Message(OP.LIST_REVISION_HISTORY).put("revisionList", revisions))); + conn.send(serializeMessage(new Message(OP.LIST_REVISION_HISTORY) + .put("noteId", noteId) + .put("revisionList", revisions))); } else { conn.send(serializeMessage( new Message(OP.ERROR_INFO).put("info", @@ -1689,7 +1695,9 @@ private void listRevisionHistory(NotebookSocket conn, @Override public void onSuccess(List revisions, ServiceContext context) throws IOException { super.onSuccess(revisions, context); - conn.send(serializeMessage(new Message(OP.LIST_REVISION_HISTORY).put("revisionList", revisions))); + conn.send(serializeMessage(new Message(OP.LIST_REVISION_HISTORY) + .put("noteId", noteId) + .put("revisionList", revisions))); } }); } @@ -1705,7 +1713,9 @@ private void setNoteRevision(NotebookSocket conn, public void onSuccess(Note note, ServiceContext context) throws IOException { super.onSuccess(note, context); Note reloadedNote = getNotebook().loadNoteFromRepo(noteId, context.getAutheInfo()); - conn.send(serializeMessage(new Message(OP.SET_NOTE_REVISION).put("status", true))); + conn.send(serializeMessage(new Message(OP.SET_NOTE_REVISION) + .put("noteId", noteId) + .put("status", true))); broadcastNote(reloadedNote); } }); diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-interpreter.interface.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-interpreter.interface.ts index c59e459410e..f9a4ac4957e 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-interpreter.interface.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-interpreter.interface.ts @@ -26,6 +26,7 @@ export interface InterpreterItem { } export interface InterpreterBindings { + noteId: string; interpreterBindings: InterpreterBindingItem[]; } diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-notebook.interface.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-notebook.interface.ts index 665e8dfd71f..2a9d11b1bdf 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-notebook.interface.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-notebook.interface.ts @@ -172,10 +172,12 @@ export interface ParagraphAdded { } export interface SetNoteRevisionStatus { + noteId: string; status: boolean; } export interface ListRevision { + noteId: string; revisionList: RevisionListItem[]; } diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts index 3dd6e739607..c0a28330a72 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts @@ -118,6 +118,11 @@ export class NotebookComponent extends MessageListenersManager implements OnInit @MessageListener(OP.INTERPRETER_BINDINGS) loadInterpreterBindings(data: MessageReceiveDataTypeMap[OP.INTERPRETER_BINDINGS]) { + const { noteId } = this.activatedRoute.snapshot.params; + if (data.noteId !== noteId) { + return; + } + this.interpreterBindings = data.interpreterBindings; if (!this.interpreterBindings.some(item => item.selected)) { this.activatedExtension = 'interpreter'; @@ -198,8 +203,11 @@ export class NotebookComponent extends MessageListenersManager implements OnInit } @MessageListener(OP.SET_NOTE_REVISION) - setNoteRevision(_data: MessageReceiveDataTypeMap[OP.SET_NOTE_REVISION]) { + setNoteRevision(data: MessageReceiveDataTypeMap[OP.SET_NOTE_REVISION]) { const { noteId } = this.activatedRoute.snapshot.params; + if (data.noteId !== noteId) { + return; + } this.router.navigate(['/notebook', noteId]).then(); } @@ -257,6 +265,11 @@ export class NotebookComponent extends MessageListenersManager implements OnInit @MessageListener(OP.LIST_REVISION_HISTORY) listRevisionHistory(data: MessageReceiveDataTypeMap[OP.LIST_REVISION_HISTORY]) { + const { noteId } = this.activatedRoute.snapshot.params; + if (data.noteId !== noteId) { + return; + } + this.noteRevisions = data.revisionList; if (this.noteRevisions) { if (this.noteRevisions.length === 0 || this.noteRevisions[0].id !== 'Head') { From 2755d9052c38aeeab655b362d5964b4914113b43 Mon Sep 17 00:00:00 2001 From: gyowoo1113 Date: Tue, 22 Sep 2026 16:51:53 +0900 Subject: [PATCH 14/15] [ZEPPELIN-6683] Add notebook reply noteId tests --- .../zeppelin/socket/NotebookServerTest.java | 92 +++++++++- .../notebook/notebook.component.spec.ts | 171 ++++++++++++++++++ 2 files changed, 257 insertions(+), 6 deletions(-) create mode 100644 zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java index d288851fbff..5de5b1d10f1 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java @@ -1067,15 +1067,17 @@ void getInterpreterBindingsRequiresReaderPermission() throws IOException { ArgumentCaptor response = ArgumentCaptor.forClass(String.class); verify(socket).send(response.capture()); - assertEquals(OP.AUTH_INFO, notebookServer.deserializeMessage(response.getValue()).op); + Message responseMessage = notebookServer.deserializeMessage(response.getValue()); + assertEquals(OP.AUTH_INFO, responseMessage.op); reset(socket); setNotePermissions(noteId, "binding-owner", "binding-reader"); notebookServer.getInterpreterBindings(socket, serviceContext("binding-reader"), message); verify(socket).send(response.capture()); - assertEquals(OP.INTERPRETER_BINDINGS, - notebookServer.deserializeMessage(response.getValue()).op); + responseMessage = notebookServer.deserializeMessage(response.getValue()); + assertEquals(OP.INTERPRETER_BINDINGS, responseMessage.op); + assertEquals(noteId, responseMessage.data.get("noteId")); } finally { notebook.removeNote(noteId, owner); } @@ -1100,7 +1102,8 @@ void saveInterpreterBindingsRequiresWriterPermission() throws IOException { notebook.processNote(noteId, Note::getDefaultInterpreterGroup)); ArgumentCaptor response = ArgumentCaptor.forClass(String.class); verify(socket).send(response.capture()); - assertEquals(OP.AUTH_INFO, notebookServer.deserializeMessage(response.getValue()).op); + Message responseMessage = notebookServer.deserializeMessage(response.getValue()); + assertEquals(OP.AUTH_INFO, responseMessage.op); reset(socket); authorizationService.setWriters(noteId, @@ -1110,8 +1113,9 @@ void saveInterpreterBindingsRequiresWriterPermission() throws IOException { assertEquals(replacementGroup, notebook.processNote(noteId, Note::getDefaultInterpreterGroup)); verify(socket).send(response.capture()); - assertEquals(OP.INTERPRETER_BINDINGS, - notebookServer.deserializeMessage(response.getValue()).op); + responseMessage = notebookServer.deserializeMessage(response.getValue()); + assertEquals(OP.INTERPRETER_BINDINGS, responseMessage.op); + assertEquals(noteId, responseMessage.data.get("noteId")); } finally { notebook.removeNote(noteId, owner); } @@ -1170,6 +1174,82 @@ void testNoteRevision() throws IOException { } } + @Test + void listRevisionHistoryIncludesNoteId() throws IOException { + String noteId = notebook.createNote("revision-list-note-id", anonymous); + + try { + NotebookSocket socket = createWebSocket(); + Message request = new Message(OP.LIST_REVISION_HISTORY) + .put("noteId", noteId); + + notebookServer.onMessage(socket, request.toJson()); + + ArgumentCaptor response = ArgumentCaptor.forClass(String.class); + verify(socket).send(response.capture()); + + Message responseMessage = notebookServer.deserializeMessage(response.getValue()); + assertEquals(OP.LIST_REVISION_HISTORY, responseMessage.op); + assertEquals(noteId, responseMessage.data.get("noteId")); + } finally { + notebook.removeNote(noteId, anonymous); + } + } + + @Test + void checkpointNoteIncludesNoteId() throws IOException { + String noteId = notebook.createNote("checkpoint-note-id", anonymous); + + try { + NotebookSocket socket = createWebSocket(); + Message request = new Message(OP.CHECKPOINT_NOTE) + .put("noteId", noteId) + .put("commitMessage", "checkpoint"); + + notebookServer.onMessage(socket, request.toJson()); + + ArgumentCaptor response = ArgumentCaptor.forClass(String.class); + verify(socket).send(response.capture()); + + Message responseMessage = notebookServer.deserializeMessage(response.getValue()); + assertEquals(OP.LIST_REVISION_HISTORY, responseMessage.op); + assertEquals(noteId, responseMessage.data.get("noteId")); + } finally { + notebook.removeNote(noteId, anonymous); + } + } + + @Test + void setNoteRevisionIncludesNoteId() throws IOException { + String noteId = notebook.createNote("set-revision-note-id", anonymous); + + try { + NotebookRepoWithVersionControl.Revision revision = + notebook.processNote(noteId, + note -> notebook.checkpointNote( + note.getId(), + note.getPath(), + "revision", + anonymous)); + + NotebookSocket socket = createWebSocket(); + Message request = new Message(OP.SET_NOTE_REVISION) + .put("noteId", noteId) + .put("revisionId", revision.id); + + notebookServer.onMessage(socket, request.toJson()); + + ArgumentCaptor response = ArgumentCaptor.forClass(String.class); + verify(socket).send(response.capture()); + + Message responseMessage = notebookServer.deserializeMessage(response.getValue()); + assertEquals(OP.SET_NOTE_REVISION, responseMessage.op); + assertEquals(noteId, responseMessage.data.get("noteId")); + } finally { + notebook.removeNote(noteId, anonymous); + } + } + private NotebookSocket createWebSocket() { NotebookSocket sock = mock(NotebookSocket.class); return sock; diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts new file mode 100644 index 00000000000..59ec120ee27 --- /dev/null +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts @@ -0,0 +1,171 @@ +/* + * 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 { NEVER } from 'rxjs'; +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import type { ChangeDetectorRef } from '@angular/core'; +import type { Title } from '@angular/platform-browser'; +import type { ActivatedRoute, Router } from '@angular/router'; +import { OP, type MessageReceiveDataTypeMap } from '@zeppelin/sdk'; + +import type { + MessageService, + NgZService, + NoteStatusService, + NoteVarShareService, + ReactFeatureService, + SecurityService, + ThemeService, + TicketService +} from '@zeppelin/services'; + +vi.mock('./paragraph/paragraph.component', () => ({ + NotebookParagraphComponent: class NotebookParagraphComponent {} +})); + +import { NotebookComponent } from './notebook.component'; + +const createComponent = () => { + const messageService = { + receive: vi.fn(() => NEVER) + } as unknown as MessageService; + + const router = { + navigate: vi.fn(() => Promise.resolve(true)) + } as unknown as Router; + + const component = new NotebookComponent( + messageService, + {} as NgZService, + { + snapshot: { + params: { + noteId: 'note-b' + } + } + } as unknown as ActivatedRoute, + { + markForCheck: vi.fn() + } as unknown as ChangeDetectorRef, + {} as NoteStatusService, + {} as NoteVarShareService, + {} as TicketService, + {} as SecurityService, + router, + {} as Title, + {} as ThemeService, + {} as ReactFeatureService + ); + + return { + component, + router + }; +}; + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe('NotebookComponent', () => { + it('ignores stale interpreter bindings replies', () => { + const { component } = createComponent(); + + const originalBindings = [{ id: 'existing', selected: true }]; + component.interpreterBindings = originalBindings as typeof component.interpreterBindings; + + const data: MessageReceiveDataTypeMap[OP.INTERPRETER_BINDINGS] = { + noteId: 'note-a', + interpreterBindings: [{ id: 'stale', selected: false }] + }; + + component.loadInterpreterBindings(data); + + expect(component.interpreterBindings).toBe(originalBindings); + }); + + it('ignores stale revision history replies', () => { + const { component } = createComponent(); + + const originalRevisions = [{ id: 'Head', message: 'Head' }]; + component.noteRevisions = originalRevisions; + component.currentRevision = 'Head'; + + const data: MessageReceiveDataTypeMap[OP.LIST_REVISION_HISTORY] = { + noteId: 'note-a', + revisionList: [{ id: 'rev-1', message: 'stale revision' }] + }; + + component.listRevisionHistory(data); + + expect(component.noteRevisions).toBe(originalRevisions); + expect(component.currentRevision).toBe('Head'); + }); + + it('ignores stale set note revision replies', () => { + const { component, router } = createComponent(); + + const data: MessageReceiveDataTypeMap[OP.SET_NOTE_REVISION] = { + noteId: 'note-a', + status: true + }; + + component.setNoteRevision(data); + + expect(router.navigate).not.toHaveBeenCalled(); + }); + + it('applies interpreter bindings for the active note', () => { + const { component } = createComponent(); + + const data: MessageReceiveDataTypeMap[OP.INTERPRETER_BINDINGS] = { + noteId: 'note-b', + interpreterBindings: [{ id: 'current', selected: true }] + }; + + component.loadInterpreterBindings(data); + + expect(component.interpreterBindings).toEqual([{ id: 'current', selected: true }]); + }); + + it('applies revision history for the active note', () => { + const { component } = createComponent(); + + const data: MessageReceiveDataTypeMap[OP.LIST_REVISION_HISTORY] = { + noteId: 'note-b', + revisionList: [{ id: 'rev-1', message: 'current revision' }] + }; + + component.listRevisionHistory(data); + + expect(component.noteRevisions).toEqual([ + { id: 'Head', message: 'Head' }, + { id: 'rev-1', message: 'current revision' } + ]); + + expect(component.currentRevision).toBe('Head'); + }); + + it('navigates after setting a revision for the active note', () => { + const { component, router } = createComponent(); + + const data: MessageReceiveDataTypeMap[OP.SET_NOTE_REVISION] = { + noteId: 'note-b', + status: true + }; + + component.setNoteRevision(data); + + expect(router.navigate).toHaveBeenCalledWith(['/notebook', 'note-b']); + }); +}); From 5deec5c62ed2265ee6c5b628256fdb903a61f0e4 Mon Sep 17 00:00:00 2001 From: gyowoo1113 Date: Tue, 22 Sep 2026 17:43:40 +0900 Subject: [PATCH 15/15] [ZEPPELIN-6683] Use envelope msgId for paragraph added messages --- .../interfaces/message-notebook.interface.ts | 1 - .../notebook/notebook.component.spec.ts | 36 ++++++++++++++++--- .../workspace/notebook/notebook.component.ts | 17 ++++++--- .../src/app/services/message.service.ts | 4 --- 4 files changed, 43 insertions(+), 15 deletions(-) diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-notebook.interface.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-notebook.interface.ts index 2a9d11b1bdf..fca5c071989 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-notebook.interface.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-notebook.interface.ts @@ -167,7 +167,6 @@ export interface ImportNoteReceived { export interface ParagraphAdded { index: number; - msgId?: string; paragraph: ParagraphItem; } diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts index 59ec120ee27..62186de41f4 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts @@ -16,7 +16,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import type { ChangeDetectorRef } from '@angular/core'; import type { Title } from '@angular/platform-browser'; import type { ActivatedRoute, Router } from '@angular/router'; -import { OP, type MessageReceiveDataTypeMap } from '@zeppelin/sdk'; +import { OP, type MessageReceiveDataTypeMap, type WebSocketMessage } from '@zeppelin/sdk'; import type { MessageService, @@ -37,15 +37,17 @@ import { NotebookComponent } from './notebook.component'; const createComponent = () => { const messageService = { - receive: vi.fn(() => NEVER) - } as unknown as MessageService; + receive: vi.fn(() => NEVER), + receiveEnvelope: vi.fn(() => NEVER), + consumeLocalAddFocusMsgId: vi.fn(() => false) + }; const router = { navigate: vi.fn(() => Promise.resolve(true)) } as unknown as Router; const component = new NotebookComponent( - messageService, + messageService as unknown as MessageService, {} as NgZService, { snapshot: { @@ -69,7 +71,8 @@ const createComponent = () => { return { component, - router + router, + messageService }; }; @@ -168,4 +171,27 @@ describe('NotebookComponent', () => { expect(router.navigate).toHaveBeenCalledWith(['/notebook', 'note-b']); }); + + it('uses the paragraph added envelope msgId for local focus', () => { + const { component, messageService } = createComponent(); + + component.note = { + paragraphs: [] + } as typeof component.note; + + const message: WebSocketMessage = { + op: OP.PARAGRAPH_ADDED, + msgId: 'local-add-msg', + data: { + index: 0, + paragraph: { + id: 'paragraph-1' + } as MessageReceiveDataTypeMap[OP.PARAGRAPH_ADDED]['paragraph'] + } + }; + + component.addParagraph(message); + + expect(messageService.consumeLocalAddFocusMsgId).toHaveBeenCalledWith('local-add-msg'); + }); }); diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts index c0a28330a72..cf05e7c0a2a 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts @@ -28,7 +28,7 @@ import { distinctUntilChanged, distinctUntilKeyChanged, startWith, takeUntil } f import { NzResizeEvent } from 'ng-zorro-antd/resizable'; -import { MessageListener, MessageListenersManager } from '@zeppelin/core'; +import { MessageEnvelopeListener, MessageListener, MessageListenersManager } from '@zeppelin/core'; import { Permissions } from '@zeppelin/interfaces'; import { DynamicFormParams, @@ -36,7 +36,8 @@ import { MessageReceiveDataTypeMap, Note, OP, - RevisionListItem + RevisionListItem, + WebSocketMessage } from '@zeppelin/sdk'; import { MessageService, @@ -151,8 +152,8 @@ export class NotebookComponent extends MessageListenersManager implements OnInit this.cdr.markForCheck(); } - @MessageListener(OP.PARAGRAPH_ADDED) - addParagraph(data: MessageReceiveDataTypeMap[OP.PARAGRAPH_ADDED]) { + @MessageEnvelopeListener(OP.PARAGRAPH_ADDED) + addParagraph(message: WebSocketMessage) { const { paragraphId } = this.activatedRoute.snapshot.params; if (paragraphId || this.revisionView) { return; @@ -160,6 +161,12 @@ export class NotebookComponent extends MessageListenersManager implements OnInit if (!this.note) { return; } + + const data = message.data; + if (data === undefined) { + return; + } + const definedNote = this.note; definedNote.paragraphs.splice(data.index, 0, data.paragraph); const paragraphIndex = definedNote.paragraphs.findIndex(p => p.id === data.paragraph.id); @@ -169,7 +176,7 @@ export class NotebookComponent extends MessageListenersManager implements OnInit // Focus the editor only for a clone/insert initiated by this client (not auto-append on run or remote inserts). // Defer a tick so the new paragraph's editor child exists, since `focus = true` alone misses it. - if (this.messageService.consumeLocalAddFocusMsgId(data.msgId)) { + if (this.messageService.consumeLocalAddFocusMsgId(message.msgId)) { const addedId = data.paragraph.id; setTimeout(() => { const added = this.listOfNotebookParagraphComponent?.find(e => e.paragraph.id === addedId); diff --git a/zeppelin-web-angular/src/app/services/message.service.ts b/zeppelin-web-angular/src/app/services/message.service.ts index 050d78c44cc..937754d400a 100644 --- a/zeppelin-web-angular/src/app/services/message.service.ts +++ b/zeppelin-web-angular/src/app/services/message.service.ts @@ -23,7 +23,6 @@ import { MessageSendDataTypeMap, Note, NoteConfig, - OP, ParagraphConfig, ParagraphParams, PersonalizedMode, @@ -52,9 +51,6 @@ export class MessageService extends Message implements OnDestroy { interceptReceived(data: WebSocketMessage): WebSocketMessage { const received = this.messageInterceptor ? this.messageInterceptor.received(data) : super.interceptReceived(data); - if (received.op === OP.PARAGRAPH_ADDED && received.data && received.msgId) { - (received.data as MessageReceiveDataTypeMap[OP.PARAGRAPH_ADDED]).msgId = received.msgId; - } return received; }