This is an automated email from the ASF dual-hosted git repository. JunRuiLee pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8ae4de695201457bb0c582426859262f46709324 Author: Martijn Visser <[email protected]> AuthorDate: Mon Jul 13 18:56:43 2026 +0200 [FLINK-31931][runtime-web] Stop polling a TaskManager once it is gone The TaskManager status and metrics views re-requested the TaskManager detail on every refresh tick, so a TaskManager that had been lost kept generating failing /taskmanagers/<id> requests indefinitely. loadManager also swallowed every error, hiding the not-found case from callers. Let loadManager propagate errors and share a small poll helper that stops once the backend reports the TaskManager is gone (404), while transient errors keep retrying. The status view then shows a "no longer available" notice instead of an endless skeleton, and the metrics cards are hidden instead of rendering blank tables. The global HTTP interceptor turns every error body into a persistent "Server Response Message" notification, which for the now-expected 404 meant a full server stack trace popping up next to the notice. loadManager marks its request with an EXPECTED_NOT_FOUND HttpContext token and the interceptor skips the notification for a 404 on such a request; every other error is still surfaced as before. Generated-by: Claude Fable 5.1 --- .../web-dashboard/src/app/app.interceptor.spec.ts | 61 ++++++++++++++++- .../web-dashboard/src/app/app.interceptor.ts | 4 +- .../metrics/task-manager-metrics.component.html | 6 +- .../metrics/task-manager-metrics.component.spec.ts | 14 ++++ .../metrics/task-manager-metrics.component.ts | 20 +++--- .../status/task-manager-status.component.html | 77 +++++++++++---------- .../status/task-manager-status.component.spec.ts | 79 ++++++++++++++++++++++ .../status/task-manager-status.component.ts | 27 ++++---- .../task-manager/task-manager-detail-poll.spec.ts | 77 +++++++++++++++++++++ .../pages/task-manager/task-manager-detail-poll.ts | 58 ++++++++++++++++ .../services/{public-api.ts => http-context.ts} | 17 +++-- .../web-dashboard/src/app/services/public-api.ts | 1 + .../src/app/services/task-manager.service.spec.ts | 21 ++++-- .../src/app/services/task-manager.service.ts | 13 ++-- 14 files changed, 391 insertions(+), 84 deletions(-) diff --git a/flink-runtime-web/web-dashboard/src/app/app.interceptor.spec.ts b/flink-runtime-web/web-dashboard/src/app/app.interceptor.spec.ts index 2b41327a0b0..957e402c3f3 100644 --- a/flink-runtime-web/web-dashboard/src/app/app.interceptor.spec.ts +++ b/flink-runtime-web/web-dashboard/src/app/app.interceptor.spec.ts @@ -16,10 +16,17 @@ * limitations under the License. */ -import { HttpErrorResponse, HttpHandler, HttpHeaders, HttpRequest, HttpResponse } from '@angular/common/http'; +import { + HttpContext, + HttpErrorResponse, + HttpHandler, + HttpHeaders, + HttpRequest, + HttpResponse +} from '@angular/common/http'; import { of, Subject, throwError } from 'rxjs'; -import { StatusService } from '@flink-runtime-web/services'; +import { EXPECTED_NOT_FOUND, StatusService } from '@flink-runtime-web/services'; import { NzNotificationService } from 'ng-zorro-antd/notification'; import { type Mock, afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; @@ -161,6 +168,56 @@ describe('AppInterceptor', () => { expect(notificationService.info).not.toHaveBeenCalled(); }); + it('surfaces a 404 with an error body by default', () => { + const error = new HttpErrorResponse({ + status: 404, + url: '/taskmanagers/tm-1', + error: { errors: ['RestHandlerException: Could not find TaskExecutor tm-1.'] } + }); + handle.mockReturnValue(throwError(() => error)); + + interceptor.intercept(new HttpRequest('GET', '/taskmanagers/tm-1'), handler).subscribe({ error: () => {} }); + + expect(statusService.listOfErrorMessage).toHaveLength(1); + expect(notificationService.info).toHaveBeenCalledTimes(1); + }); + + it('does not surface an expected 404 but still re-throws it to the caller', () => { + const error = new HttpErrorResponse({ + status: 404, + url: '/taskmanagers/tm-1', + error: { errors: ['RestHandlerException: Could not find TaskExecutor tm-1.'] } + }); + handle.mockReturnValue(throwError(() => error)); + const request = new HttpRequest('GET', '/taskmanagers/tm-1', { + context: new HttpContext().set(EXPECTED_NOT_FOUND, true) + }); + + let caught: unknown; + interceptor.intercept(request, handler).subscribe({ error: err => (caught = err) }); + + expect(caught).toBe(error); + expect(statusService.listOfErrorMessage).toEqual([]); + expect(notificationService.info).not.toHaveBeenCalled(); + }); + + it('still surfaces other errors on a request that expects a 404', () => { + const error = new HttpErrorResponse({ + status: 500, + url: '/taskmanagers/tm-1', + error: { errors: ['Internal server error.'] } + }); + handle.mockReturnValue(throwError(() => error)); + const request = new HttpRequest('GET', '/taskmanagers/tm-1', { + context: new HttpContext().set(EXPECTED_NOT_FOUND, true) + }); + + interceptor.intercept(request, handler).subscribe({ error: () => {} }); + + expect(statusService.listOfErrorMessage).toEqual(['Internal server error.']); + expect(notificationService.info).toHaveBeenCalledTimes(1); + }); + it.each([0, 500, 503])( 'counts a bodyless status %i as a network failure without surfacing a warning below the threshold', status => { diff --git a/flink-runtime-web/web-dashboard/src/app/app.interceptor.ts b/flink-runtime-web/web-dashboard/src/app/app.interceptor.ts index c77b890da33..dc13cb10538 100644 --- a/flink-runtime-web/web-dashboard/src/app/app.interceptor.ts +++ b/flink-runtime-web/web-dashboard/src/app/app.interceptor.ts @@ -29,7 +29,7 @@ import { Injectable } from '@angular/core'; import { Observable, throwError } from 'rxjs'; import { catchError, tap } from 'rxjs/operators'; -import { StatusService } from '@flink-runtime-web/services'; +import { EXPECTED_NOT_FOUND, StatusService } from '@flink-runtime-web/services'; import { NzNotificationService, NzNotificationDataOptions } from 'ng-zorro-antd/notification'; @Injectable() @@ -71,8 +71,10 @@ export class AppInterceptor implements HttpInterceptor { } const errorMessage = res && res.error && res.error.errors && res.error.errors[0]; + const expectedNotFound = res.status === HttpStatusCode.NotFound && req.context.get(EXPECTED_NOT_FOUND); if ( errorMessage && + !expectedNotFound && ignoreErrorUrlEndsList.every(url => !res.url.endsWith(url)) && ignoreErrorMessage.every(message => errorMessage !== message) ) { diff --git a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/metrics/task-manager-metrics.component.html b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/metrics/task-manager-metrics.component.html index fdc16301270..6c59658a39a 100644 --- a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/metrics/task-manager-metrics.component.html +++ b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/metrics/task-manager-metrics.component.html @@ -16,7 +16,7 @@ ~ limitations under the License. --> -<nz-card nzTitle="Memory" nzSize="small" class="flink-memory-model"> +<nz-card *ngIf="taskManagerDetail" nzTitle="Memory" nzSize="small" class="flink-memory-model"> <nz-table nzBordered *ngIf="metrics" @@ -164,7 +164,7 @@ </tbody> </nz-table> </nz-card> -<nz-card nzTitle="Advanced" nzSize="small" class="flink-memory-model"> +<nz-card *ngIf="taskManagerDetail" nzTitle="Advanced" nzSize="small" class="flink-memory-model"> <div nz-row [nzGutter]="16"> <div nz-col [nzSpan]="12"> <nz-table @@ -318,7 +318,7 @@ </tr> </ng-template> -<nz-card nzTitle="Resources" nzSize="small" class="flink-memory-model"> +<nz-card *ngIf="taskManagerDetail" nzTitle="Resources" nzSize="small" class="flink-memory-model"> <nz-table nzBordered nzTitle="Unassigned resources" diff --git a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/metrics/task-manager-metrics.component.spec.ts b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/metrics/task-manager-metrics.component.spec.ts index b95645bad30..0eebdd6296b 100644 --- a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/metrics/task-manager-metrics.component.spec.ts +++ b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/metrics/task-manager-metrics.component.spec.ts @@ -16,6 +16,7 @@ * limitations under the License. */ +import { HttpErrorResponse } from '@angular/common/http'; import { ComponentFixture, TestBed } from '@angular/core/testing'; import { ActivatedRoute } from '@angular/router'; import { of, throwError } from 'rxjs'; @@ -114,4 +115,17 @@ describe('TaskManagerMetricsComponent', () => { expect(fixture.componentInstance.taskManagerDetail).toBeUndefined(); expect(loadMetrics).not.toHaveBeenCalled(); }); + + it('hides the metric cards (no NaN) when the TaskManager is gone (404)', () => { + loadManager.mockReturnValue(throwError(() => new HttpErrorResponse({ status: 404 }))); + + fixture.detectChanges(); + + expect(fixture.componentInstance.taskManagerDetail).toBeUndefined(); + expect(loadMetrics).not.toHaveBeenCalled(); + const text = element.textContent ?? ''; + expect(text).not.toContain('Flink Memory Model'); + expect(text).not.toContain('Advanced'); + expect(text).not.toContain('NaN'); + }); }); diff --git a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/metrics/task-manager-metrics.component.ts b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/metrics/task-manager-metrics.component.ts index c7c4350aa23..a3498f68caf 100644 --- a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/metrics/task-manager-metrics.component.ts +++ b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/metrics/task-manager-metrics.component.ts @@ -20,7 +20,7 @@ import { DecimalPipe, NgForOf, NgIf, NgTemplateOutlet } from '@angular/common'; import { ChangeDetectionStrategy, ChangeDetectorRef, Component, OnDestroy, OnInit } from '@angular/core'; import { ActivatedRoute } from '@angular/router'; import { of, Subject } from 'rxjs'; -import { catchError, mergeMap, startWith, takeUntil } from 'rxjs/operators'; +import { catchError, takeUntil } from 'rxjs/operators'; import { HumanizeBytesPipe } from '@flink-runtime-web/components/humanize-bytes.pipe'; import { MetricMap, TaskManagerDetail } from '@flink-runtime-web/interfaces'; @@ -32,6 +32,8 @@ import { NzProgressModule } from 'ng-zorro-antd/progress'; import { NzTableModule } from 'ng-zorro-antd/table'; import { NzTooltipModule } from 'ng-zorro-antd/tooltip'; +import { pollTaskManagerDetail } from '../task-manager-detail-poll'; + @Component({ selector: 'flink-task-manager-metrics', templateUrl: './task-manager-metrics.component.html', @@ -66,17 +68,13 @@ export class TaskManagerMetricsComponent implements OnInit, OnDestroy { public ngOnInit(): void { const taskManagerId = this.activatedRoute.parent!.snapshot.params.taskManagerId; - this.statusService.refresh$ - .pipe( - startWith(true), - mergeMap(() => this.taskManagerService.loadManager(taskManagerId).pipe(catchError(() => of(undefined)))), - takeUntil(this.destroy$) - ) - .subscribe(data => { - if (data) { - this.reload(data.id); + pollTaskManagerDetail(this.statusService.refresh$, () => this.taskManagerService.loadManager(taskManagerId)) + .pipe(takeUntil(this.destroy$)) + .subscribe(({ detail }) => { + if (detail) { + this.reload(detail.id); } - this.taskManagerDetail = data; + this.taskManagerDetail = detail; this.cdr.markForCheck(); }); } diff --git a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/status/task-manager-status.component.html b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/status/task-manager-status.component.html index 46e954d7082..8c4be8f6f93 100644 --- a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/status/task-manager-status.component.html +++ b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/status/task-manager-status.component.html @@ -17,40 +17,49 @@ --> <ng-container *ngIf="!loading; else loadingContent"> - <div class="title-wrapper" [attr.title]="taskManagerDetail?.id"> - <span class="title">{{ taskManagerDetail?.id }}</span> - <flink-blocked-badge *ngIf="taskManagerDetail?.blocked"></flink-blocked-badge> - </div> - <nz-descriptions *ngIf="taskManagerDetail" nzBordered nzSize="small"> - <nz-descriptions-item [nzSpan]="1" nzTitle="Path"> - {{ taskManagerDetail.path }} - </nz-descriptions-item> - <nz-descriptions-item [nzSpan]="1" nzTitle="Free/All Slots"> - {{ taskManagerDetail.freeSlots }} / {{ taskManagerDetail.slotsNumber }} - </nz-descriptions-item> - <nz-descriptions-item [nzSpan]="1" nzTitle="Assigned Tasks"> - {{ taskManagerDetail.assignedTasks }} - </nz-descriptions-item> - <nz-descriptions-item [nzSpan]="1" nzTitle="Last Heartbeat"> - {{ taskManagerDetail.timeSinceLastHeartbeat | date: 'yyyy-MM-dd HH:mm:ss' }} - </nz-descriptions-item> - <nz-descriptions-item [nzSpan]="1" nzTitle="Data Port"> - {{ taskManagerDetail.dataPort }} - </nz-descriptions-item> - <nz-descriptions-item [nzSpan]="1" nzTitle="CPU Cores"> - {{ taskManagerDetail.hardware.cpuCores }} - </nz-descriptions-item> - <nz-descriptions-item [nzSpan]="1" nzTitle="Physical Memory"> - {{ taskManagerDetail.hardware.physicalMemory | humanizeBytes }} - </nz-descriptions-item> - <nz-descriptions-item [nzSpan]="1" nzTitle="JVM Heap Size"> - {{ taskManagerDetail.hardware.freeMemory | humanizeBytes }} - </nz-descriptions-item> - <nz-descriptions-item [nzSpan]="1" nzTitle="Flink Managed Memory"> - {{ taskManagerDetail.hardware.managedMemory | humanizeBytes }} - </nz-descriptions-item> - </nz-descriptions> - <flink-navigation [listOfNavigation]="listOfNavigation"></flink-navigation> + <nz-alert + *ngIf="notFound; else detailContent" + nzType="warning" + nzShowIcon + nzMessage="TaskManager not available" + nzDescription="This TaskManager is no longer registered. It may have been shut down or lost." + ></nz-alert> + <ng-template #detailContent> + <div class="title-wrapper" [attr.title]="taskManagerDetail?.id"> + <span class="title">{{ taskManagerDetail?.id }}</span> + <flink-blocked-badge *ngIf="taskManagerDetail?.blocked"></flink-blocked-badge> + </div> + <nz-descriptions *ngIf="taskManagerDetail" nzBordered nzSize="small"> + <nz-descriptions-item [nzSpan]="1" nzTitle="Path"> + {{ taskManagerDetail.path }} + </nz-descriptions-item> + <nz-descriptions-item [nzSpan]="1" nzTitle="Free/All Slots"> + {{ taskManagerDetail.freeSlots }} / {{ taskManagerDetail.slotsNumber }} + </nz-descriptions-item> + <nz-descriptions-item [nzSpan]="1" nzTitle="Assigned Tasks"> + {{ taskManagerDetail.assignedTasks }} + </nz-descriptions-item> + <nz-descriptions-item [nzSpan]="1" nzTitle="Last Heartbeat"> + {{ taskManagerDetail.timeSinceLastHeartbeat | date: 'yyyy-MM-dd HH:mm:ss' }} + </nz-descriptions-item> + <nz-descriptions-item [nzSpan]="1" nzTitle="Data Port"> + {{ taskManagerDetail.dataPort }} + </nz-descriptions-item> + <nz-descriptions-item [nzSpan]="1" nzTitle="CPU Cores"> + {{ taskManagerDetail.hardware.cpuCores }} + </nz-descriptions-item> + <nz-descriptions-item [nzSpan]="1" nzTitle="Physical Memory"> + {{ taskManagerDetail.hardware.physicalMemory | humanizeBytes }} + </nz-descriptions-item> + <nz-descriptions-item [nzSpan]="1" nzTitle="JVM Heap Size"> + {{ taskManagerDetail.hardware.freeMemory | humanizeBytes }} + </nz-descriptions-item> + <nz-descriptions-item [nzSpan]="1" nzTitle="Flink Managed Memory"> + {{ taskManagerDetail.hardware.managedMemory | humanizeBytes }} + </nz-descriptions-item> + </nz-descriptions> + <flink-navigation [listOfNavigation]="listOfNavigation"></flink-navigation> + </ng-template> </ng-container> <ng-template #loadingContent> diff --git a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/status/task-manager-status.component.spec.ts b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/status/task-manager-status.component.spec.ts new file mode 100644 index 00000000000..1f49dbb0ae4 --- /dev/null +++ b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/status/task-manager-status.component.spec.ts @@ -0,0 +1,79 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 { HttpErrorResponse } from '@angular/common/http'; +import { ComponentFixture, TestBed } from '@angular/core/testing'; +import { ActivatedRoute } from '@angular/router'; +import { of, throwError } from 'rxjs'; + +import { TaskManagerDetail } from '@flink-runtime-web/interfaces'; +import { StatusService, TaskManagerService } from '@flink-runtime-web/services'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { TaskManagerStatusComponent } from './task-manager-status.component'; + +const mockDetail = { + id: 'tm-container-7', + path: 'pekko.tcp://[email protected]:6122/user/rpc/taskmanager', + dataPort: 43210, + timeSinceLastHeartbeat: 1_781_000_000_000, + slotsNumber: 4, + freeSlots: 2, + assignedTasks: 2, + hardware: { cpuCores: 8, physicalMemory: 16_000_000_000, freeMemory: 8_000_000_000, managedMemory: 4_000_000_000 }, + blocked: false +} as unknown as TaskManagerDetail; + +describe('TaskManagerStatusComponent', () => { + let fixture: ComponentFixture<TaskManagerStatusComponent>; + let element: HTMLElement; + const loadManager = vi.fn(); + + beforeEach(async () => { + loadManager.mockReset().mockReturnValue(of(mockDetail)); + await TestBed.configureTestingModule({ + imports: [TaskManagerStatusComponent], + providers: [ + { provide: StatusService, useValue: { refresh$: of(true) } }, + { provide: TaskManagerService, useValue: { loadManager } }, + { provide: ActivatedRoute, useValue: { snapshot: { params: { taskManagerId: 'tm-container-7' } } } } + ] + }).compileComponents(); + fixture = TestBed.createComponent(TaskManagerStatusComponent); + element = fixture.nativeElement as HTMLElement; + }); + + it('renders the TaskManager detail when it is available', () => { + fixture.detectChanges(); + + expect(fixture.componentInstance.notFound).toBe(false); + expect(element.textContent).toContain('tm-container-7'); + expect(element.textContent).toContain('43210'); + expect(element.textContent).not.toContain('no longer registered'); + }); + + it('shows a not-available notice when the TaskManager is gone (404)', () => { + loadManager.mockReturnValue(throwError(() => new HttpErrorResponse({ status: 404 }))); + + fixture.detectChanges(); + + expect(fixture.componentInstance.notFound).toBe(true); + expect(fixture.componentInstance.taskManagerDetail).toBeUndefined(); + expect(element.textContent).toContain('TaskManager not available'); + }); +}); diff --git a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/status/task-manager-status.component.ts b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/status/task-manager-status.component.ts index f579f7e518e..b3c2569cf09 100644 --- a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/status/task-manager-status.component.ts +++ b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/status/task-manager-status.component.ts @@ -19,17 +19,20 @@ import { DatePipe, NgIf } from '@angular/common'; import { ChangeDetectionStrategy, ChangeDetectorRef, Component, OnDestroy, OnInit } from '@angular/core'; import { ActivatedRoute } from '@angular/router'; -import { of, Subject } from 'rxjs'; -import { catchError, mergeMap, takeUntil } from 'rxjs/operators'; +import { Subject } from 'rxjs'; +import { takeUntil } from 'rxjs/operators'; import { BlockedBadgeComponent } from '@flink-runtime-web/components/blocked-badge/blocked-badge.component'; import { HumanizeBytesPipe } from '@flink-runtime-web/components/humanize-bytes.pipe'; import { NavigationComponent } from '@flink-runtime-web/components/navigation/navigation.component'; import { TaskManagerDetail } from '@flink-runtime-web/interfaces'; import { StatusService, TaskManagerService } from '@flink-runtime-web/services'; +import { NzAlertModule } from 'ng-zorro-antd/alert'; import { NzDescriptionsModule } from 'ng-zorro-antd/descriptions'; import { NzSkeletonModule } from 'ng-zorro-antd/skeleton'; +import { pollTaskManagerDetail } from '../task-manager-detail-poll'; + @Component({ selector: 'flink-task-manager-status', templateUrl: './task-manager-status.component.html', @@ -38,6 +41,7 @@ import { NzSkeletonModule } from 'ng-zorro-antd/skeleton'; imports: [ NgIf, BlockedBadgeComponent, + NzAlertModule, NzDescriptionsModule, DatePipe, HumanizeBytesPipe, @@ -56,6 +60,7 @@ export class TaskManagerStatusComponent implements OnInit, OnDestroy { ]; public taskManagerDetail?: TaskManagerDetail; public loading = true; + public notFound = false; private readonly destroy$ = new Subject<void>(); @@ -67,17 +72,13 @@ export class TaskManagerStatusComponent implements OnInit, OnDestroy { ) {} public ngOnInit(): void { - this.statusService.refresh$ - .pipe( - mergeMap(() => - this.taskManagerService - .loadManager(this.activatedRoute.snapshot.params.taskManagerId) - .pipe(catchError(() => of(undefined))) - ), - takeUntil(this.destroy$) - ) - .subscribe(data => { - this.taskManagerDetail = data; + pollTaskManagerDetail(this.statusService.refresh$, () => + this.taskManagerService.loadManager(this.activatedRoute.snapshot.params.taskManagerId) + ) + .pipe(takeUntil(this.destroy$)) + .subscribe(({ detail, notFound }) => { + this.taskManagerDetail = detail; + this.notFound = notFound; this.loading = false; this.cdr.markForCheck(); }); diff --git a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/task-manager-detail-poll.spec.ts b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/task-manager-detail-poll.spec.ts new file mode 100644 index 00000000000..b9947befd0d --- /dev/null +++ b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/task-manager-detail-poll.spec.ts @@ -0,0 +1,77 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 { HttpErrorResponse } from '@angular/common/http'; +import { Subject, of, throwError } from 'rxjs'; + +import { TaskManagerDetail } from '@flink-runtime-web/interfaces'; +import { describe, expect, it, vi } from 'vitest'; + +import { pollTaskManagerDetail, TaskManagerDetailResult } from './task-manager-detail-poll'; + +const detail = { id: 'tm-1' } as unknown as TaskManagerDetail; + +describe('pollTaskManagerDetail', () => { + it('emits the detail on every successful tick', () => { + const tick$ = new Subject<void>(); + const results: TaskManagerDetailResult[] = []; + pollTaskManagerDetail(tick$, () => of(detail)).subscribe(result => results.push(result)); + + tick$.next(); + tick$.next(); + + expect(results).toEqual([ + { detail, notFound: false }, + { detail, notFound: false } + ]); + }); + + it('reports notFound and stops polling once the TaskManager is gone (404)', () => { + const tick$ = new Subject<void>(); + const load = vi.fn(() => throwError(() => new HttpErrorResponse({ status: 404 }))); + const results: TaskManagerDetailResult[] = []; + let completed = false; + pollTaskManagerDetail(tick$, load).subscribe({ + next: result => results.push(result), + complete: () => (completed = true) + }); + + tick$.next(); + tick$.next(); + + expect(results).toEqual([{ detail: undefined, notFound: true }]); + expect(load).toHaveBeenCalledTimes(1); + expect(completed).toBe(true); + }); + + it('ignores a transient (non-404) error and keeps polling', () => { + const tick$ = new Subject<void>(); + const load = vi + .fn() + .mockReturnValueOnce(throwError(() => new HttpErrorResponse({ status: 503 }))) + .mockReturnValueOnce(of(detail)); + const results: TaskManagerDetailResult[] = []; + pollTaskManagerDetail(tick$, load).subscribe(result => results.push(result)); + + tick$.next(); + tick$.next(); + + expect(results).toEqual([{ detail, notFound: false }]); + expect(load).toHaveBeenCalledTimes(2); + }); +}); diff --git a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/task-manager-detail-poll.ts b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/task-manager-detail-poll.ts new file mode 100644 index 00000000000..7635e8e0c26 --- /dev/null +++ b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/task-manager-detail-poll.ts @@ -0,0 +1,58 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 { HttpErrorResponse } from '@angular/common/http'; +import { defer, EMPTY, Observable, of } from 'rxjs'; +import { catchError, map, switchMap, takeWhile } from 'rxjs/operators'; + +import { TaskManagerDetail } from '@flink-runtime-web/interfaces'; + +/** Outcome of polling a single TaskManager's detail. */ +export interface TaskManagerDetailResult { + detail?: TaskManagerDetail; + notFound: boolean; +} + +/** + * Loads a single TaskManager's detail on every tick, stopping for good once the backend reports + * the TaskManager is gone (404). A gone TaskManager never comes back (its id is unique per + * registration), so re-polling it is futile; transient errors are ignored and keep polling. + */ +export function pollTaskManagerDetail( + tick$: Observable<unknown>, + load: () => Observable<TaskManagerDetail> +): Observable<TaskManagerDetailResult> { + return defer(() => { + let notFound = false; + return tick$.pipe( + takeWhile(() => !notFound), + switchMap(() => + load().pipe( + map(detail => ({ detail, notFound: false })), + catchError((error: unknown) => { + if (error instanceof HttpErrorResponse && error.status === 404) { + notFound = true; + return of({ detail: undefined, notFound: true }); + } + return EMPTY; + }) + ) + ) + ); + }); +} diff --git a/flink-runtime-web/web-dashboard/src/app/services/public-api.ts b/flink-runtime-web/web-dashboard/src/app/services/http-context.ts similarity index 69% copy from flink-runtime-web/web-dashboard/src/app/services/public-api.ts copy to flink-runtime-web/web-dashboard/src/app/services/http-context.ts index df66d9dcaba..ff13e34c082 100644 --- a/flink-runtime-web/web-dashboard/src/app/services/public-api.ts +++ b/flink-runtime-web/web-dashboard/src/app/services/http-context.ts @@ -16,12 +16,11 @@ * limitations under the License. */ -export * from './status.service'; -export * from './overview.service'; -export * from './job.service'; -export * from './jar.service'; -export * from './job-manager.service'; -export * from './task-manager.service'; -export * from './metrics.service'; -export * from './config.service'; -export * from './application.service'; +import { HttpContextToken } from '@angular/common/http'; + +/** + * Marks a request for which a 404 is an expected outcome that the caller handles itself, so the + * global interceptor does not surface it as a server error notification. Any other error is + * still surfaced as usual. + */ +export const EXPECTED_NOT_FOUND = new HttpContextToken<boolean>(() => false); diff --git a/flink-runtime-web/web-dashboard/src/app/services/public-api.ts b/flink-runtime-web/web-dashboard/src/app/services/public-api.ts index df66d9dcaba..6031f10afcb 100644 --- a/flink-runtime-web/web-dashboard/src/app/services/public-api.ts +++ b/flink-runtime-web/web-dashboard/src/app/services/public-api.ts @@ -17,6 +17,7 @@ */ export * from './status.service'; +export * from './http-context'; export * from './overview.service'; export * from './job.service'; export * from './jar.service'; diff --git a/flink-runtime-web/web-dashboard/src/app/services/task-manager.service.spec.ts b/flink-runtime-web/web-dashboard/src/app/services/task-manager.service.spec.ts index dc73c3d1d85..e95d49fb082 100644 --- a/flink-runtime-web/web-dashboard/src/app/services/task-manager.service.spec.ts +++ b/flink-runtime-web/web-dashboard/src/app/services/task-manager.service.spec.ts @@ -16,13 +16,14 @@ * limitations under the License. */ -import { HttpClient } from '@angular/common/http'; -import { firstValueFrom, of, throwError, toArray } from 'rxjs'; +import { HttpClient, HttpContext } from '@angular/common/http'; +import { firstValueFrom, of, throwError } from 'rxjs'; import { TaskManagerDetail, TaskManagersItem } from '@flink-runtime-web/interfaces'; import { beforeEach, describe, expect, it, vi } from 'vitest'; import { ConfigService } from './config.service'; +import { EXPECTED_NOT_FOUND } from './http-context'; import { TaskManagerService } from './task-manager.service'; describe('TaskManagerService', () => { @@ -71,12 +72,20 @@ describe('TaskManagerService', () => { expect(result).toBe(detail); }); - it('swallows request errors into an empty stream, unlike loadManagers which falls back to []', async () => { - httpClient.get.mockReturnValue(throwError(() => new Error('cluster unreachable'))); + it('propagates request errors so callers can react to a gone TaskManager, unlike loadManagers', async () => { + const error = new Error('cluster unreachable'); + httpClient.get.mockReturnValue(throwError(() => error)); + + await expect(firstValueFrom(service.loadManager('tm-1'))).rejects.toBe(error); + }); + + it('marks the request as expecting a 404 so the interceptor does not surface it', () => { + httpClient.get.mockReturnValue(of({ id: 'tm-1' })); - const emissions = await firstValueFrom(service.loadManager('tm-1').pipe(toArray())); + service.loadManager('tm-1'); - expect(emissions).toEqual([]); + const [, options] = httpClient.get.mock.calls[0] as [string, { context: HttpContext }]; + expect(options.context.get(EXPECTED_NOT_FOUND)).toBe(true); }); }); }); diff --git a/flink-runtime-web/web-dashboard/src/app/services/task-manager.service.ts b/flink-runtime-web/web-dashboard/src/app/services/task-manager.service.ts index 3d3ff96172c..8a810b71e1a 100644 --- a/flink-runtime-web/web-dashboard/src/app/services/task-manager.service.ts +++ b/flink-runtime-web/web-dashboard/src/app/services/task-manager.service.ts @@ -16,9 +16,9 @@ * limitations under the License. */ -import { HttpClient, HttpHeaders } from '@angular/common/http'; +import { HttpClient, HttpContext, HttpHeaders } from '@angular/common/http'; import { Injectable } from '@angular/core'; -import { EMPTY, Observable, of } from 'rxjs'; +import { Observable, of } from 'rxjs'; import { catchError, map } from 'rxjs/operators'; import { @@ -34,6 +34,7 @@ import { import { ProfilingDetail, ProfilingList } from '@flink-runtime-web/interfaces/job-profiler'; import { ConfigService } from './config.service'; +import { EXPECTED_NOT_FOUND } from './http-context'; @Injectable({ providedIn: 'root' @@ -52,9 +53,11 @@ export class TaskManagerService { } loadManager(taskManagerId: string): Observable<TaskManagerDetail> { - return this.httpClient - .get<TaskManagerDetail>(`${this.configService.BASE_URL}/taskmanagers/${taskManagerId}`) - .pipe(catchError(() => EMPTY)); + // Let errors propagate (e.g. a 404 for a gone TaskManager) so callers can react to them. The 404 + // is expected and handled by the callers, so it is not surfaced as a server error notification. + return this.httpClient.get<TaskManagerDetail>(`${this.configService.BASE_URL}/taskmanagers/${taskManagerId}`, { + context: new HttpContext().set(EXPECTED_NOT_FOUND, true) + }); } loadLogList(taskManagerId: string): Observable<TaskManagerLogItem[]> {
