From 04dc3063a571f5bbd02945fa1dc3d492b05c99a2 Mon Sep 17 00:00:00 2001 From: HwangRock Date: Sat, 15 Aug 2026 20:02:19 +0900 Subject: [PATCH 1/3] [ZEPPELIN-5235] Extract Note.abortAll() and share it with REST stopNoteJobs --- .../org/apache/zeppelin/notebook/Note.java | 11 +++++++ .../apache/zeppelin/rest/NotebookRestApi.java | 6 +--- .../zeppelin/notebook/NotebookTest.java | 30 +++++++++++++++++++ 3 files changed, 42 insertions(+), 5 deletions(-) diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Note.java b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Note.java index 278fa43ecc6..ccebad12dc3 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Note.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Note.java @@ -815,6 +815,17 @@ public List getParagraphs() { return this.paragraphs; } + /** + * Abort all the paragraphs which are not terminated yet. + */ + public void abortAll() { + for (Paragraph p : getParagraphs()) { + if (!p.isTerminated()) { + p.abort(); + } + } + } + // TODO(zjffdu) how does this used ? private void snapshotAngularObjectRegistry(String user) { angularObjects = new HashMap<>(); diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java index 192cd5056c5..3c09a612f42 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java @@ -887,11 +887,7 @@ public Response stopNoteJobs(@PathParam("noteId") String noteId) note -> { checkIfNoteIsNotNull(note, noteId); checkIfUserCanRun(noteId, "Insufficient privileges you cannot stop this job for this note"); - for (Paragraph p : note.getParagraphs()) { - if (!p.isTerminated()) { - p.abort(); - } - } + note.abortAll(); return new JsonResponse<>(Status.OK).build(); }); } diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java index 39e6ec70e38..a9a879795e1 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java @@ -616,6 +616,36 @@ void testRunAll() throws Exception { notebook.removeNote(noteId, anonymous); } + @Test + void testAbortAll() throws IOException { + String noteId = notebook.createNote("note1", anonymous); + notebook.processNote(noteId, + note -> { + Paragraph p1 = note.addNewParagraph(AuthenticationInfo.ANONYMOUS); + p1.setText("p1"); + p1.setStatus(Status.RUNNING); + + Paragraph p2 = note.addNewParagraph(AuthenticationInfo.ANONYMOUS); + p2.setText("p2"); + p2.setStatus(Status.PENDING); + + Paragraph p3 = note.addNewParagraph(AuthenticationInfo.ANONYMOUS); + p3.setText("p3"); + p3.setStatus(Status.FINISHED); + + // when + note.abortAll(); + + // then + assertTrue(p1.isAborted()); + assertTrue(p2.isAborted()); + assertFalse(p3.isAborted()); + assertEquals(Status.FINISHED, p3.getStatus()); + return null; + }); + notebook.removeNote(noteId, anonymous); + } + @Test void testSchedule() throws InterruptedException, IOException { // create a note and a paragraph From 30b836327e1062decf365d389a3013be8fd0984c Mon Sep 17 00:00:00 2001 From: HwangRock Date: Sat, 15 Aug 2026 20:02:43 +0900 Subject: [PATCH 2/3] [ZEPPELIN-5235] Add CANCEL_ALL_PARAGRAPHS WS OP and NotebookService support --- .../org/apache/zeppelin/common/Message.java | 1 + .../zeppelin/service/NotebookService.java | 20 ++++++ .../zeppelin/socket/NotebookServer.java | 8 +++ .../zeppelin/service/NotebookServiceTest.java | 61 +++++++++++++++++++ .../zeppelin/socket/NotebookServerTest.java | 46 ++++++++++++++ 5 files changed, 136 insertions(+) diff --git a/zeppelin-common/src/main/java/org/apache/zeppelin/common/Message.java b/zeppelin-common/src/main/java/org/apache/zeppelin/common/Message.java index 6ec66e63dbd..126855bcffa 100644 --- a/zeppelin-common/src/main/java/org/apache/zeppelin/common/Message.java +++ b/zeppelin-common/src/main/java/org/apache/zeppelin/common/Message.java @@ -192,6 +192,7 @@ public enum OP { PARAGRAPH_MOVED, // [s-c] paragraph moved NOTE_UPDATED, // [s-c] paragraph updated(name, config) RUN_ALL_PARAGRAPHS, // [c-s] run all paragraphs + CANCEL_ALL_PARAGRAPHS, // [c-s] cancel(abort) all paragraphs PARAGRAPH_EXECUTED_BY_SPELL, // [c-s] paragraph was executed by spell RUN_PARAGRAPH_USING_SPELL, // [s-c] run paragraph using spell PARAS_INFO, // [s-c] paragraph runtime infos diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java b/zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java index 38ce280ad6e..9e5e31aa1ce 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java @@ -602,6 +602,26 @@ public void cancelParagraph(String noteId, } + public void cancelAllParagraphs(String noteId, + ServiceContext context, + ServiceCallback callback) throws IOException { + if (!checkPermission(noteId, Permission.RUNNER, Message.OP.CANCEL_ALL_PARAGRAPHS, context, + callback)) { + return; + } + + notebook.processNote(noteId, + note -> { + if (note == null) { + throw new NoteNotFoundException(noteId); + } + note.abortAll(); + callback.onSuccess(null, context); + return null; + }); + + } + public void moveParagraph(String noteId, String paragraphId, int newIndex, 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 2d78ae7fb57..555d22ffd38 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 @@ -387,6 +387,9 @@ public void onMessage(NotebookSocket conn, String msg) { case CANCEL_PARAGRAPH: cancelParagraph(conn, context, receivedMessage); break; + case CANCEL_ALL_PARAGRAPHS: + cancelAllParagraphs(conn, context, receivedMessage); + break; case MOVE_PARAGRAPH: moveParagraph(conn, context, receivedMessage); break; @@ -1482,6 +1485,11 @@ private void cancelParagraph(NotebookSocket conn, ServiceContext context, Messag getNotebookService().cancelParagraph(noteId, paragraphId, context, new WebSocketServiceCallback<>(conn)); } + private void cancelAllParagraphs(NotebookSocket conn, ServiceContext context, Message fromMessage) throws IOException { + final String noteId = (String) fromMessage.get("noteId"); + getNotebookService().cancelAllParagraphs(noteId, context, new WebSocketServiceCallback<>(conn)); + } + private void runAllParagraphs(NotebookSocket conn, ServiceContext context, Message fromMessage) throws IOException { diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java index 2d53f34dec8..2eeb0f650c6 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java @@ -19,8 +19,10 @@ package org.apache.zeppelin.service; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; import static org.mockito.ArgumentMatchers.any; @@ -67,6 +69,9 @@ import org.apache.zeppelin.notebook.repo.NotebookRepo; import org.apache.zeppelin.notebook.repo.VFSNotebookRepo; import org.apache.zeppelin.notebook.scheduler.QuartzSchedulerService; +import org.apache.zeppelin.rest.exception.ForbiddenException; +import org.apache.zeppelin.rest.exception.NoteNotFoundException; +import org.apache.zeppelin.scheduler.Job.Status; import org.apache.zeppelin.search.LuceneSearch; import org.apache.zeppelin.search.SearchService; import org.apache.zeppelin.storage.ConfigStorage; @@ -671,6 +676,62 @@ void testRunParagraphInPersonalizedModeDoesNotPolluteMasterParagraph() throws IO }); } + @Test + void testCancelAllParagraphs() throws IOException { + String note1Id = notebookService.createNote("note_cancel_all", "python", false, context, callback); + Paragraph p1 = notebook.processNote(note1Id, + note1 -> { + Paragraph p = note1.addNewParagraph(context.getAutheInfo()); + p.setText("p1"); + p.setStatus(Status.RUNNING); + return p; + }); + Paragraph p2 = notebook.processNote(note1Id, + note1 -> { + Paragraph p = note1.addNewParagraph(context.getAutheInfo()); + p.setText("p2"); + p.setStatus(Status.FINISHED); + return p; + }); + + reset(callback); + notebookService.cancelAllParagraphs(note1Id, context, callback); + + assertTrue(p1.isAborted()); + assertFalse(p2.isAborted()); + verify(callback).onSuccess(any(), eq(context)); + } + + @Test + void testCancelAllParagraphsForbidden() throws IOException { + String note1Id = notebookService.createNote("note_cancel_all_forbidden", "python", false, context, callback); + Paragraph p1 = notebook.processNote(note1Id, + note1 -> { + Paragraph p = note1.addNewParagraph(context.getAutheInfo()); + p.setText("p1"); + p.setStatus(Status.RUNNING); + return p; + }); + + HashSet otherUser = new HashSet<>(); + otherUser.add("other_user"); + authorizationService.setOwners(note1Id, otherUser); + authorizationService.setWriters(note1Id, otherUser); + authorizationService.setRunners(note1Id, otherUser); + + reset(callback); + notebookService.cancelAllParagraphs(note1Id, context, callback); + + assertFalse(p1.isAborted()); + verify(callback).onFailure(any(ForbiddenException.class), eq(context)); + } + + @Test + void testCancelAllParagraphsNoteNotFound() { + assertThrows(NoteNotFoundException.class, + () -> notebookService.cancelAllParagraphs("non_existing_note_id", context, callback)); + } + @Test void testNormalizeNotePath() throws IOException { assertEquals("/Untitled Note", notebookService.normalizeNotePath(" ")); 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 d982d46a33c..ef6eb6d5ec0 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 @@ -237,6 +237,52 @@ void testOnParagraphStatusChange_whenJobManagerDisabled() { } } + @Test + void testCancelAllParagraphsNotDisabledForRunningNotes() { + assertFalse(Message.isDisabledForRunningNotes(OP.CANCEL_ALL_PARAGRAPHS)); + } + + @Test + void testCancelAllParagraphsWebSocket() throws IOException { + NotebookSocket sock1 = createWebSocket(); + + String noteName = "Note with millis " + System.currentTimeMillis(); + notebookServer.onMessage(sock1, new Message(OP.NEW_NOTE).put("name", noteName).toJson()); + NoteInfo createdNoteInfo = null; + for (NoteInfo noteInfo : notebook.getNotesInfo()) { + if (notebook.processNote(noteInfo.getId(), Note::getName).equals(noteName)) { + createdNoteInfo = noteInfo; + break; + } + } + String noteId = createdNoteInfo.getId(); + + notebookServer.onMessage(sock1, new Message(OP.GET_NOTE).put("id", noteId).toJson()); + + Paragraph paragraph = notebook.processNote(noteId, + note -> { + Paragraph p = note.getParagraphs().get(0); + p.setStatus(Status.RUNNING); + // simulate a sequential run in progress + note.setRunning(true); + return p; + }); + + try { + notebookServer.onMessage(sock1, + new Message(OP.CANCEL_ALL_PARAGRAPHS).put("noteId", noteId).toJson()); + + assertTrue(paragraph.isAborted()); + } finally { + notebook.processNote(noteId, + note -> { + note.setRunning(false); + return null; + }); + notebook.removeNote(noteId, anonymous); + } + } + @Test void testCollaborativeEditing() throws IOException { if (!zepServer.getZeppelinConfiguration().isZeppelinNotebookCollaborativeModeEnable()) { From c3399523c213383389e36d340bc704fd369897d1 Mon Sep 17 00:00:00 2001 From: HwangRock Date: Sat, 15 Aug 2026 20:02:57 +0900 Subject: [PATCH 3/3] [ZEPPELIN-5235] Add Cancel all button to New UI action bar --- .../e2e/models/notebook-action-bar-page.ts | 2 ++ .../action-bar/action-bar-functionality.spec.ts | 8 ++++++++ .../src/interfaces/message-data-type-map.interface.ts | 2 ++ .../src/interfaces/message-operator.interface.ts | 6 ++++++ .../src/interfaces/message-paragraph.interface.ts | 4 ++++ .../projects/zeppelin-sdk/src/message.ts | 4 ++++ .../notebook/action-bar/action-bar.component.html | 11 +++++++++++ .../notebook/action-bar/action-bar.component.ts | 4 ++++ .../src/app/services/message.service.ts | 4 ++++ 9 files changed, 45 insertions(+) diff --git a/zeppelin-web-angular/e2e/models/notebook-action-bar-page.ts b/zeppelin-web-angular/e2e/models/notebook-action-bar-page.ts index 1ecc33bd2c1..3ffe207727f 100644 --- a/zeppelin-web-angular/e2e/models/notebook-action-bar-page.ts +++ b/zeppelin-web-angular/e2e/models/notebook-action-bar-page.ts @@ -16,6 +16,7 @@ import { BasePage } from './base-page'; export class NotebookActionBarPage extends BasePage { readonly titleEditor: Locator; readonly runAllButton: Locator; + readonly cancelAllButton: Locator; readonly showHideCodeButton: Locator; readonly showHideOutputButton: Locator; readonly clearOutputButton: Locator; @@ -40,6 +41,7 @@ export class NotebookActionBarPage extends BasePage { super(page); this.titleEditor = page.locator('zeppelin-elastic-input'); this.runAllButton = page.locator('button[nzTooltipTitle="Run all paragraphs"]'); + this.cancelAllButton = page.locator('button[nzTooltipTitle="Cancel all paragraphs"]'); this.showHideCodeButton = page.locator('button[nzTooltipTitle="Show/hide the code"]'); this.showHideOutputButton = page.locator('button[nzTooltipTitle="Show/hide the output"]'); this.clearOutputButton = page.locator('button[nzTooltipTitle="Clear all output"]'); diff --git a/zeppelin-web-angular/e2e/tests/notebook/action-bar/action-bar-functionality.spec.ts b/zeppelin-web-angular/e2e/tests/notebook/action-bar/action-bar-functionality.spec.ts index 5825e1d31b1..0ef046d3c14 100644 --- a/zeppelin-web-angular/e2e/tests/notebook/action-bar/action-bar-functionality.spec.ts +++ b/zeppelin-web-angular/e2e/tests/notebook/action-bar/action-bar-functionality.spec.ts @@ -68,6 +68,14 @@ test.describe('Notebook Action Bar Functionality', () => { await expect(confirmButton).not.toBeVisible(); }); + test('should display cancel all button as disabled when note is idle', async () => { + await expect(actionBarPage.cancelAllButton).toBeVisible(); + + // Given: an idle note (no paragraph running), Cancel all is disabled and Run all is enabled — the two buttons are mutually exclusive + await expect(actionBarPage.cancelAllButton).toBeDisabled(); + await expect(actionBarPage.runAllButton).toBeEnabled(); + }); + test('should toggle code visibility', async () => { await expect(actionBarPage.showHideCodeButton).toBeVisible(); await expect(actionBarPage.showHideCodeButton).toBeEnabled(); 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 6c6088c73ae..5fba20a1326 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 @@ -59,6 +59,7 @@ import { AngularObjectRemove, AngularObjectUpdate, AngularObjectUpdated, + CancelAllParagraphs, CancelParagraph, CommitParagraph, Completion, @@ -154,6 +155,7 @@ export interface MessageSendDataTypeMap { [OP.PARAGRAPH_EXECUTED_BY_SPELL]: ParagraphExecutedBySpell; [OP.RUN_PARAGRAPH]: RunParagraph; [OP.RUN_ALL_PARAGRAPHS]: RunAllParagraphs; + [OP.CANCEL_ALL_PARAGRAPHS]: CancelAllParagraphs; [OP.PARAGRAPH_REMOVE]: ParagraphRemove; [OP.PARAGRAPH_CLEAR_OUTPUT]: ParagraphClearOutput; [OP.PARAGRAPH_CLEAR_ALL_OUTPUT]: ParagraphClearAllOutput; diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-operator.interface.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-operator.interface.ts index 1f8036b3931..322fb8f3880 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-operator.interface.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-operator.interface.ts @@ -443,6 +443,12 @@ export enum OP { */ RUN_ALL_PARAGRAPHS = 'RUN_ALL_PARAGRAPHS', + /** + * [c-s] + * cancel all paragraphs + */ + CANCEL_ALL_PARAGRAPHS = 'CANCEL_ALL_PARAGRAPHS', + /** * [c-s] * paragraph was executed by spell 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 2ea3916ea1e..f75cd1f5f31 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 @@ -178,6 +178,10 @@ export interface RunAllParagraphs { paragraphs: string; } +export interface CancelAllParagraphs { + noteId: string; +} + export interface InsertParagraph { index: number; } diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts index 6262bff26a4..110af36f27b 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts @@ -415,6 +415,10 @@ export class Message { }); } + cancelAllParagraphs(noteId: string): void { + this.send(OP.CANCEL_ALL_PARAGRAPHS, { noteId }); + } + paragraphRemove(paragraphId: string): void { this.send(OP.PARAGRAPH_REMOVE, { id: paragraphId }); } diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/action-bar/action-bar.component.html b/zeppelin-web-angular/src/app/pages/workspace/notebook/action-bar/action-bar.component.html index f53f5292dae..c45438e8724 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/action-bar/action-bar.component.html +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/action-bar/action-bar.component.html @@ -33,6 +33,17 @@ > + @if (!viewOnly) {