This is an automated email from the ASF dual-hosted git repository.
voidmatcha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/zeppelin.git
The following commit(s) were added to refs/heads/master by this push:
new 2ad49b3d6c [ZEPPELIN-6551] Fix Job Manager crash on removal broadcast
for a note not in the viewer's list
2ad49b3d6c is described below
commit 2ad49b3d6c7b510af9a69f2e85b8ed47795c02ab
Author: 김예나 <[email protected]>
AuthorDate: Mon Jul 27 22:04:54 2026 +0900
[ZEPPELIN-6551] Fix Job Manager crash on removal broadcast for a note not
in the viewer's list
### What is this PR for?
When a note is permanently deleted, the backend broadcasts a field-less
`NoteJobInfo` stub (`noteName=null`, `isRemoved=true`) to **every** Job Manager
subscriber, while each viewer's initial job list is owner-filtered
(`JobManagerService#getNoteJobInfoByUnixTime`).
For a note the viewer does not own, `updateJobs` fell into the
`currentJobIndex === -1` branch and pushed the stub into `this.jobs`.
`filterJobs` then ran `job.noteName.match(noteNameReg)` and threw `TypeError:
Cannot read properties of undefined (reading 'match')`. Because the stub
persists in `this.jobs`, every subsequent `filterJobs` (each filter keystroke /
job update) rethrew, so the page's filter/sort stayed broken until
re-navigation.
This affects any user viewing `/jobmanager` when another user permanently
deletes a note they don't own.
### What does this PR do?
- `updateJobs`: do not add removal stubs that aren't already in the list
(`if (!updateJob.isRemoved)`). The guard is kept nested inside `if
(currentJobIndex === -1)` rather than merged into that condition, so a removal
stub can never fall through to the `else` branch's `splice(currentJobIndex, 1)`
— merging would call `splice(-1, 1)` and silently drop the last job.
- `filterJobs`: guard `job.noteName?.match(...)` with optional chaining as
defense in depth.
### What type of PR is it?
Bug Fix
### What is the Jira issue?
* https://issues.apache.org/jira/browse/ZEPPELIN-6551
### How should this be tested?
* `cd zeppelin-web-angular && npm run lint`
* With two users: open `/jobmanager` as user A; as user B permanently
delete a note A does not own; confirm A's page keeps filtering without a
console `TypeError`.
### Questions:
* Does the license files need to update? No
* Is there breaking changes for older versions? No
* Does this needs documentation? No
Closes #5337 from kimyenac/ZEPPELIN-6551.
Signed-off-by: YONGJAE LEE <[email protected]>
---
.../e2e/models/job-manager-page.ts | 44 ++++++
.../e2e/models/job-manager-page.util.ts | 147 +++++++++++++++++++++
.../job-manager-removal-broadcast.spec.ts | 114 ++++++++++++++++
.../workspace/job-manager/job-manager.component.ts | 6 +-
4 files changed, 309 insertions(+), 2 deletions(-)
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 0000000000..7274723a2b
--- /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<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('');
+ }
+}
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 0000000000..076a190794
--- /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<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([]);
+};
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 0000000000..12fb3d49c2
--- /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 497a1c2a31..a7a9728bc2 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));
