From efcd6792a9121ec96fad4aced8eaaf2817286ce4 Mon Sep 17 00:00:00 2001 From: kimyenac Date: Fri, 24 Jul 2026 17:08:33 +0900 Subject: [PATCH 1/3] [ZEPPELIN-6551] Fix Job Manager crash on removal broadcast for a note not in the viewer's list A permanent note deletion broadcasts a field-less NoteJobInfo stub (noteName=null, isRemoved=true) to every Job Manager subscriber, while each viewer's job list is owner-filtered. For a note the viewer doesn't own, updateJobs pushed the stub (currentJobIndex === -1), then filterJobs threw "Cannot read properties of undefined (reading 'match')" on job.noteName and kept rethrowing, breaking filter/sort until re-navigation. - updateJobs: don't add removal stubs that aren't already in the list. Guard is nested (not merged into the currentJobIndex === -1 condition) so a removal stub can't fall through to the splice(-1) branch. - filterJobs: guard job.noteName with optional chaining as defense in depth. --- .../pages/workspace/job-manager/job-manager.component.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/zeppelin-web-angular/src/app/pages/workspace/job-manager/job-manager.component.ts b/zeppelin-web-angular/src/app/pages/workspace/job-manager/job-manager.component.ts index 497a1c2a31f..a7a9728bc28 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/job-manager/job-manager.component.ts +++ b/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)); From 3686ba4304234a23b19aa9e9045eeb1618844853 Mon Sep 17 00:00:00 2001 From: kimyenac Date: Sat, 25 Jul 2026 17:41:42 +0900 Subject: [PATCH 2/3] [ZEPPELIN-6551] Add e2e regression tests for the Job Manager removal broadcast Cover the removal broadcast that `NotebookServer#onNoteRemove` sends to every Job Manager subscriber: a removal stub for a note absent from the viewer's list must be ignored, must not break the note name filter, and must not drop a listed job. The Job Manager's WebSocket traffic is stubbed because `zeppelin.jobmanager.enable` defaults to false and the backend list is owner-filtered, so the cross-user broadcast cannot be produced from a single session. All other WebSocket traffic still goes to the real server. --- .../e2e/models/job-manager-page.ts | 44 ++++++ .../e2e/models/job-manager-page.util.ts | 147 ++++++++++++++++++ .../job-manager-removal-broadcast.spec.ts | 101 ++++++++++++ 3 files changed, 292 insertions(+) create mode 100644 zeppelin-web-angular/e2e/models/job-manager-page.ts create mode 100644 zeppelin-web-angular/e2e/models/job-manager-page.util.ts create mode 100644 zeppelin-web-angular/e2e/tests/workspace/job-manager/job-manager-removal-broadcast.spec.ts diff --git a/zeppelin-web-angular/e2e/models/job-manager-page.ts b/zeppelin-web-angular/e2e/models/job-manager-page.ts new file mode 100644 index 00000000000..7274723a2b6 --- /dev/null +++ b/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 { + 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 { + await this.fillAndVerifyInput(this.searchInput, noteName); + } + + async clearNoteNameFilter(): Promise { + await this.searchInput.fill(''); + } +} diff --git a/zeppelin-web-angular/e2e/models/job-manager-page.util.ts b/zeppelin-web-angular/e2e/models/job-manager-page.util.ts new file mode 100644 index 00000000000..076a1907947 --- /dev/null +++ b/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 { + 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([]); +}; diff --git a/zeppelin-web-angular/e2e/tests/workspace/job-manager/job-manager-removal-broadcast.spec.ts b/zeppelin-web-angular/e2e/tests/workspace/job-manager/job-manager-removal-broadcast.spec.ts new file mode 100644 index 00000000000..cf1db55a582 --- /dev/null +++ b/zeppelin-web-angular/e2e/tests/workspace/job-manager/job-manager-removal-broadcast.spec.ts @@ -0,0 +1,101 @@ +/* + * 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 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); + }); + + 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); + + // The stub carries no `noteName`, so rendering it would throw; it must not reach the list. + await expect(jobManagerPage.jobItems).toHaveCount(2); + 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); + await expect(jobManagerPage.jobItems).toHaveCount(2); + + // 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); + + // 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.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(); + }); +}); From 8007d2247c91dfccfca6149e393b57520c28efd5 Mon Sep 17 00:00:00 2001 From: kimyenac Date: Sun, 26 Jul 2026 14:10:54 +0900 Subject: [PATCH 3/3] [ZEPPELIN-6551] Make the removal broadcast e2e assertions fail on a regression Ignoring a removal stub leaves the job list untouched, so the assertions that ran right after the broadcast were already satisfied by the pre-broadcast state, and a regressed build freezes the rendered list in that same state, so they could not fail. With the fix reverted the previous spec still passed the "no listed job is dropped" test outright, and the runtime-error test passed or failed depending on when the pageerror happened to land. Broadcast a sentinel job after the removal and wait for it to render instead. Both messages travel the same socket in FIFO order into the same listener, so the sentinel appearing proves the removal was handled first; on a regressed build filterJobs keeps throwing and it never appears. The note name filter test already had that barrier via the search box, so only its redundant count assertion is dropped. Reverting the fix now fails 3 of the 5 tests, and so does folding the removal guard into the currentJobIndex === -1 condition. --- .../job-manager-removal-broadcast.spec.ts | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/zeppelin-web-angular/e2e/tests/workspace/job-manager/job-manager-removal-broadcast.spec.ts b/zeppelin-web-angular/e2e/tests/workspace/job-manager/job-manager-removal-broadcast.spec.ts index cf1db55a582..12fb3d49c2f 100644 --- a/zeppelin-web-angular/e2e/tests/workspace/job-manager/job-manager-removal-broadcast.spec.ts +++ b/zeppelin-web-angular/e2e/tests/workspace/job-manager/job-manager-removal-broadcast.spec.ts @@ -24,6 +24,7 @@ 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 @@ -51,17 +52,27 @@ test.describe('Job Manager removal broadcast', () => { 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. - await expect(jobManagerPage.jobItems).toHaveCount(2); + // 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); - await expect(jobManagerPage.jobItems).toHaveCount(2); // Before the fix the appended stub made every later `filterJobs` throw, so the rendered // list froze and stopped reacting to the search box. @@ -76,10 +87,12 @@ test.describe('Job Manager removal broadcast', () => { 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(); });