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..12fb3d49c2f --- /dev/null +++ b/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(); + }); +}); 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));