[ZEPPELIN-5235] Add Cancel all paragraphs button to the new UI

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

zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Note.java

@@ -815,6 +815,17 @@ public List<Paragraph> 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<>();

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();
       });
   }

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<Paragraph> 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,

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 {

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

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<String> 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(" "));

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()) {

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"]');

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();

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;

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

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;
 }

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>(OP.CANCEL_ALL_PARAGRAPHS, { noteId });
+  }
+
   paragraphRemove(paragraphId: string): void {
     this.send<OP.PARAGRAPH_REMOVE>(OP.PARAGRAPH_REMOVE, { id: paragraphId });
   }

zeppelin-web-angular/src/app/pages/workspace/notebook/action-bar/action-bar.component.html

@@ -33,6 +33,17 @@
       >
         <i nz-icon nzType="play-circle" nzTheme="outline"></i>
       </button>
+      <button
+        nz-button
+        nz-popconfirm
+        nzPopconfirmTitle="Cancel all paragraphs?"
+        nz-tooltip
+        nzTooltipTitle="Cancel all paragraphs"
+        (nzOnConfirm)="cancelAllParagraphs()"
+        [disabled]="revisionView || !isNoteParagraphRunning"
+      >
+        <i nz-icon nzType="pause-circle" nzTheme="outline"></i>
+      </button>
       @if (!viewOnly) {
         <button
           nz-button

zeppelin-web-angular/src/app/pages/workspace/notebook/action-bar/action-bar.component.ts

@@ -154,6 +154,10 @@ export class NotebookActionBarComponent extends MessageListenersManager implemen
     );
   }
 
+  cancelAllParagraphs() {
+    this.messageService.cancelAllParagraphs(this.note.id);
+  }
+
   clearAllParagraphOutput() {
     this.messageService.paragraphClearAllOutput(this.note.id);
   }

zeppelin-web-angular/src/app/services/message.service.ts

@@ -277,6 +277,10 @@ export class MessageService extends Message implements OnDestroy {
     super.runAllParagraphs(noteId, paragraphs);
   }
 
+  cancelAllParagraphs(noteId: string): void {
+    super.cancelAllParagraphs(noteId);
+  }
+
   paragraphRemove(paragraphId: string): void {
     super.paragraphRemove(paragraphId);
   }