[ZEPPELIN-6551] Fix Job Manager crash on removal broadcast for a note not in the viewer's list

zeppelin-web-angular/e2e/models/job-manager-page.ts

@@ -0,0 +1,44 @@
+/*
+ * 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 { Locator, Page } from '@playwright/test';
+import { waitForZeppelinReady } from '../utils';
+import { BasePage } from './base-page';
+
+export class JobManagerPage extends BasePage {
+  readonly searchInput: Locator;
+  readonly jobItems: Locator;
+
+  constructor(page: Page) {
+    super(page);
+    this.searchInput = page.locator('input[placeholder="Search jobs..."]');
+    this.jobItems = page.locator('zeppelin-job-manager-job');
+  }
+
+  async navigate(): Promise<void> {
+    await this.navigateToRoute('/jobmanager', { timeout: 60000 });
+    await this.page.waitForURL('**/#/jobmanager', { timeout: 60000 });
+    await waitForZeppelinReady(this.page);
+  }
+
+  jobItemByName(noteName: string): Locator {
+    return this.jobItems.filter({ hasText: noteName });
+  }
+
+  async filterByNoteName(noteName: string): Promise<void> {
+    await this.fillAndVerifyInput(this.searchInput, noteName);
+  }
+
+  async clearNoteNameFilter(): Promise<void> {
+    await this.searchInput.fill('');
+  }
+}

zeppelin-web-angular/e2e/models/job-manager-page.util.ts

@@ -0,0 +1,147 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { expect, Page, WebSocketRoute } from '@playwright/test';
+
+// Zeppelin's own WebSocket endpoint; the Angular CLI serves its HMR socket on /ng-cli-ws.
+const ZEPPELIN_WS_URL_PATTERN = /\/ws(\?|$)/;
+
+const JOB_MANAGER_OPS = ['LIST_NOTE_JOBS', 'LIST_UPDATE_NOTE_JOBS', 'JOB_MANAGER_DISABLED'];
+
+// Any fixed instant works — the Job Manager only sorts by it, and no test here asserts order.
+const FIXED_UNIX_TIME = 1700000000000;
+
+interface ParagraphJobPayload {
+  id: string;
+  name: string;
+  status: string;
+}
+
+// Mirrors JobManagerService.NoteJobInfo(Note) as serialized onto the wire.
+export interface NoteJobPayload {
+  noteId: string;
+  noteName: string;
+  noteType: string;
+  interpreter: string;
+  isRunningJob: boolean;
+  isRemoved: boolean;
+  unixTimeLastRun: number;
+  paragraphs: ParagraphJobPayload[];
+}
+
+// What NotebookServer#onNoteRemove broadcasts: NoteJobInfo(noteId, isRemoved) with Gson
+// dropping the null noteName / noteType / interpreter / paragraphs fields.
+interface RemovedNoteJobPayload {
+  noteId: string;
+  isRunningJob: boolean;
+  isRemoved: true;
+  unixTimeLastRun: number;
+}
+
+type NoteJobsPayload = NoteJobPayload | RemovedNoteJobPayload;
+
+export const buildNoteJob = (noteName: string): NoteJobPayload => ({
+  noteId: `noteId_${noteName}`,
+  noteName,
+  noteType: 'normal',
+  interpreter: 'spark',
+  isRunningJob: false,
+  isRemoved: false,
+  unixTimeLastRun: FIXED_UNIX_TIME,
+  paragraphs: [{ id: `paragraph_${noteName}`, name: 'p1', status: 'FINISHED' }]
+});
+
+// Serves the Job Manager's WebSocket traffic from the test: zeppelin.jobmanager.enable
+// defaults to false and the backend list is owner-filtered, so the real feed can neither
+// guarantee a populated list nor produce a cross-user removal broadcast. Everything that is
+// not Job Manager traffic is forwarded untouched.
+export class JobManagerSocketStub {
+  private pageSocket: WebSocketRoute | null = null;
+
+  constructor(private readonly initialJobs: NoteJobPayload[]) {}
+
+  async install(page: Page): Promise<void> {
+    await page.routeWebSocket(ZEPPELIN_WS_URL_PATTERN, socket => {
+      const server = socket.connectToServer();
+
+      socket.onMessage(message => {
+        server.send(message);
+        if (opOf(message) === 'LIST_NOTE_JOBS') {
+          // The Job Manager page just subscribed — this is Zeppelin's socket, not another
+          // WebSocket that happens to live under /ws.
+          this.pageSocket = socket;
+          socket.send(listNoteJobsMessage(this.initialJobs));
+        }
+      });
+
+      server.onMessage(message => {
+        // Drop the backend's Job Manager traffic; the stub owns this page's job list.
+        const op = opOf(message);
+        if (op && JOB_MANAGER_OPS.includes(op)) {
+          return;
+        }
+        socket.send(message);
+      });
+    });
+  }
+
+  broadcastUpdate(jobs: NoteJobsPayload[]): void {
+    if (!this.pageSocket) {
+      throw new Error('JobManagerSocketStub: the Job Manager has not subscribed to the WebSocket yet');
+    }
+    this.pageSocket.send(
+      JSON.stringify({
+        op: 'LIST_UPDATE_NOTE_JOBS',
+        data: { noteRunningJobs: { lastResponseUnixTime: FIXED_UNIX_TIME, jobs } }
+      })
+    );
+  }
+
+  // The field-less stub broadcast when a note is permanently deleted.
+  broadcastRemoval(noteId: string): void {
+    this.broadcastUpdate([{ noteId, isRunningJob: false, isRemoved: true, unixTimeLastRun: 0 }]);
+  }
+}
+
+const listNoteJobsMessage = (jobs: NoteJobPayload[]): string =>
+  JSON.stringify({
+    op: 'LIST_NOTE_JOBS',
+    data: { noteJobs: { lastResponseUnixTime: FIXED_UNIX_TIME, jobs } }
+  });
+
+const opOf = (message: string | Buffer): string | undefined => {
+  try {
+    return (JSON.parse(message.toString()) as { op?: string }).op;
+  } catch {
+    return undefined;
+  }
+};
+
+export const collectRuntimeErrors = (page: Page): string[] => {
+  const errors: string[] = [];
+  page.on('pageerror', error => errors.push(error.message));
+  page.on('console', message => {
+    if (message.type() === 'error') {
+      errors.push(message.text());
+    }
+  });
+  return errors;
+};
+
+// The TypeError a removal stub used to raise inside filterJobs, worded per engine: Chromium
+// "Cannot read properties of undefined (reading 'match')", Firefox "job.noteName is
+// undefined", WebKit "undefined is not an object (evaluating 'job.noteName.match')".
+const NOTE_NAME_ACCESS_ERROR = /reading 'match'|noteName is (undefined|null)|noteName\.match/;
+
+export const expectNoNoteNameAccessError = (errors: string[]): void => {
+  expect(errors.filter(error => NOTE_NAME_ACCESS_ERROR.test(error))).toEqual([]);
+};

zeppelin-web-angular/e2e/tests/workspace/job-manager/job-manager-removal-broadcast.spec.ts

@@ -0,0 +1,114 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { expect, test } from '@playwright/test';
+import { JobManagerPage } from 'e2e/models/job-manager-page';
+import {
+  buildNoteJob,
+  collectRuntimeErrors,
+  expectNoNoteNameAccessError,
+  JobManagerSocketStub,
+  NoteJobPayload
+} from 'e2e/models/job-manager-page.util';
+import { addPageAnnotationBeforeEach, PAGES } from '../../../utils';
+
+const ALPHA_NOTE_NAME = 'JobManagerRemovalAlpha';
+const BETA_NOTE_NAME = 'JobManagerRemovalBeta';
+const GAMMA_NOTE_NAME = 'JobManagerRemovalGamma';
+const SENTINEL_NOTE_NAME = 'JobManagerRemovalSentinel';
+const UNLISTED_NOTE_ID = 'notOwnedByThisViewer';
+
+// ZEPPELIN-6551: deleting a note broadcasts a field-less removal stub to every Job Manager
+// subscriber, while each viewer's own list is owner-filtered. For a note the viewer does not
+// own the stub matched nothing in the list, got appended anyway, and filterJobs then threw on
+// its missing noteName, leaving the page's filter and sort broken until re-navigation.
+test.describe('Job Manager removal broadcast', () => {
+  addPageAnnotationBeforeEach(PAGES.WORKSPACE.JOB_MANAGER);
+
+  let jobManagerPage: JobManagerPage;
+  let socketStub: JobManagerSocketStub;
+  let runtimeErrors: string[];
+  let alphaJob: NoteJobPayload;
+
+  test.beforeEach(async ({ page }) => {
+    runtimeErrors = collectRuntimeErrors(page);
+
+    alphaJob = buildNoteJob(ALPHA_NOTE_NAME);
+    socketStub = new JobManagerSocketStub([alphaJob, buildNoteJob(BETA_NOTE_NAME)]);
+    await socketStub.install(page);
+
+    jobManagerPage = new JobManagerPage(page);
+    await jobManagerPage.navigate();
+
+    await expect(jobManagerPage.jobItems).toHaveCount(2);
+  });
+
+  // A removal stub for an unlisted note is a no-op, so asserting right after the broadcast passes on
+  // the pre-broadcast state, and a regressed build freezes the list in that same state, so such an
+  // assertion can never fail. A job sent afterwards rides the same socket FIFO, so its appearance
+  // proves the removal ran first; on a regressed build `filterJobs` throws and it never appears.
+  const awaitRemovalProcessed = async () => {
+    socketStub.broadcastUpdate([buildNoteJob(SENTINEL_NOTE_NAME)]);
+    await expect(jobManagerPage.jobItemByName(SENTINEL_NOTE_NAME)).toBeVisible();
+  };
+
+  test('Given a note absent from the list When its removal is broadcast Then no runtime error is raised', async () => {
+    socketStub.broadcastRemoval(UNLISTED_NOTE_ID);
+    await awaitRemovalProcessed();
+
+    // The stub carries no `noteName`, so rendering it would throw; it must not reach the list.
+    // Only the sentinel joins the two jobs the viewer already had.
+    await expect(jobManagerPage.jobItems).toHaveCount(3);
+    expectNoNoteNameAccessError(runtimeErrors);
+  });
+
+  test('Given a note absent from the list When its removal is broadcast Then the note name filter keeps working', async () => {
+    socketStub.broadcastRemoval(UNLISTED_NOTE_ID);
+
+    // Before the fix the appended stub made every later `filterJobs` throw, so the rendered
+    // list froze and stopped reacting to the search box.
+    await jobManagerPage.filterByNoteName(ALPHA_NOTE_NAME);
+    await expect(jobManagerPage.jobItems).toHaveCount(1);
+    await expect(jobManagerPage.jobItemByName(ALPHA_NOTE_NAME)).toBeVisible();
+
+    await jobManagerPage.clearNoteNameFilter();
+    await expect(jobManagerPage.jobItems).toHaveCount(2);
+    expectNoNoteNameAccessError(runtimeErrors);
+  });
+
+  test('Given a note absent from the list When its removal is broadcast Then no listed job is dropped', async () => {
+    socketStub.broadcastRemoval(UNLISTED_NOTE_ID);
+    await awaitRemovalProcessed();
+
+    // Guarding the removal inside the `currentJobIndex === -1` branch matters: folding the
+    // guard into that condition would send the stub to the `else` branch and have it
+    // `splice(-1, 1)` the last job out of the list.
+    await expect(jobManagerPage.jobItems).toHaveCount(3);
+    await expect(jobManagerPage.jobItemByName(ALPHA_NOTE_NAME)).toBeVisible();
+    await expect(jobManagerPage.jobItemByName(BETA_NOTE_NAME)).toBeVisible();
+  });
+
+  test('Given a note present in the list When its removal is broadcast Then only that job is removed', async () => {
+    socketStub.broadcastRemoval(alphaJob.noteId);
+
+    await expect(jobManagerPage.jobItems).toHaveCount(1);
+    await expect(jobManagerPage.jobItemByName(BETA_NOTE_NAME)).toBeVisible();
+  });
+
+  test('Given a note absent from the list When a running update is broadcast Then the job is added', async () => {
+    // Only removal stubs are dropped; updates for notes the viewer has not seen yet still join the list.
+    socketStub.broadcastUpdate([{ ...buildNoteJob(GAMMA_NOTE_NAME), isRunningJob: true }]);
+
+    await expect(jobManagerPage.jobItems).toHaveCount(3);
+    await expect(jobManagerPage.jobItemByName(GAMMA_NOTE_NAME)).toBeVisible();
+  });
+});

zeppelin-web-angular/src/app/pages/workspace/job-manager/job-manager.component.ts

@@ -61,7 +61,9 @@ export class JobManagerComponent extends MessageListenersManager implements OnDe
     data.noteRunningJobs.jobs.forEach(updateJob => {
       const currentJobIndex = this.jobs.findIndex(job => job.noteId === updateJob.noteId);
       if (currentJobIndex === -1) {
-        this.jobs.push(updateJob);
+        if (!updateJob.isRemoved) {
+          this.jobs.push(updateJob);
+        }
       } else {
         if (updateJob.isRemoved) {
           this.jobs.splice(currentJobIndex, 1);
@@ -89,7 +91,7 @@ export class JobManagerComponent extends MessageListenersManager implements OnDe
         const noteNameReg = new RegExp(escapedString, 'gi');
         return (
           (filterData.interpreter === '*' || job.interpreter === filterData.interpreter) &&
-          job.noteName.match(noteNameReg)
+          job.noteName?.match(noteNameReg)
         );
       })
       .sort((x, y) => (isSortByAsc ? x.unixTimeLastRun - y.unixTimeLastRun : y.unixTimeLastRun - x.unixTimeLastRun));