From 7a128569b60d182cb5da1cfd0433b22535a7e0dc Mon Sep 17 00:00:00 2001 From: Martijn Visser <2989614+MartijnVisser@users.noreply.github.com> Date: Mon, 13 Jul 2026 18:56:32 +0200 Subject: [PATCH 1/2] [FLINK-31931][runtime/rest] Return 404 instead of 500 for unknown TaskManager in TaskManagerDetailsHandler TaskManagerDetailsHandler stripped the failure from its exceptionally stage with stripExecutionException, which does not unwrap the CompletionException that a CompletableFuture reports failures with. The instanceof UnknownTaskExecutorException check therefore never matched, so a request for a gone TaskManager returned 500 with an "Unhandled exception" ERROR log instead of the intended 404. Use stripCompletionException, as the sibling TaskManager handlers already do. Generated-by: Claude Opus 4.8 (1M context) --- .../TaskManagerDetailsHandler.java | 2 +- .../TaskManagerDetailsHandlerTest.java | 27 +++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerDetailsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerDetailsHandler.java index 7acf165eeabaa6..a0c88c4c86ddb0 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerDetailsHandler.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerDetailsHandler.java @@ -119,7 +119,7 @@ protected CompletableFuture handleRequest( .exceptionally( (Throwable throwable) -> { final Throwable strippedThrowable = - ExceptionUtils.stripExecutionException(throwable); + ExceptionUtils.stripCompletionException(throwable); if (strippedThrowable instanceof UnknownTaskExecutorException) { throw new CompletionException( diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerDetailsHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerDetailsHandlerTest.java index 5f3825b252bfc4..c4ec69eb6b8432 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerDetailsHandlerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerDetailsHandlerTest.java @@ -24,6 +24,7 @@ import org.apache.flink.runtime.metrics.dump.MetricDump; import org.apache.flink.runtime.metrics.dump.QueryScopeInfo; import org.apache.flink.runtime.resourcemanager.TaskManagerInfoWithSlots; +import org.apache.flink.runtime.resourcemanager.exceptions.UnknownTaskExecutorException; import org.apache.flink.runtime.resourcemanager.utils.TestingResourceManagerGateway; import org.apache.flink.runtime.rest.handler.HandlerRequest; import org.apache.flink.runtime.rest.handler.HandlerRequestException; @@ -39,10 +40,12 @@ import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerMetricsInfo; import org.apache.flink.runtime.taskexecutor.TaskExecutorMemoryConfiguration; import org.apache.flink.testutils.TestingUtils; +import org.apache.flink.util.concurrent.FutureUtils; import org.apache.flink.util.jackson.JacksonMapperFactory; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -55,6 +58,7 @@ import java.util.concurrent.ExecutionException; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests the {@link TaskManagerDetailsHandler} implementation. */ class TaskManagerDetailsHandlerTest { @@ -129,6 +133,29 @@ void testTaskManagerMetricsInfoExtraction() assertThat(actualJson).isEqualTo(expectedJson); } + @Test + void testUnknownTaskExecutorLeadsToNotFound() { + resourceManagerGateway.setRequestTaskManagerDetailsInfoFunction( + taskManagerId -> + FutureUtils.completedExceptionally( + new UnknownTaskExecutorException(taskManagerId))); + + assertThatThrownBy( + () -> + testInstance + .handleRequest(createRequest(), resourceManagerGateway) + .get()) + .cause() + .isInstanceOfSatisfying( + RestHandlerException.class, + restHandlerException -> { + assertThat(restHandlerException.getHttpResponseStatus()) + .isEqualTo(HttpResponseStatus.NOT_FOUND); + assertThat(restHandlerException.getMessage()) + .contains("Could not find TaskExecutor " + TASK_MANAGER_ID); + }); + } + private static void initializeMetricStore(MetricStore metricStore) { QueryScopeInfo.TaskManagerQueryScopeInfo tmScope = new QueryScopeInfo.TaskManagerQueryScopeInfo(TASK_MANAGER_ID.toString(), "Status"); From fde2d9e389a8823bffd229383e7dd452b57059fb Mon Sep 17 00:00:00 2001 From: Martijn Visser <2989614+MartijnVisser@users.noreply.github.com> Date: Mon, 13 Jul 2026 18:56:43 +0200 Subject: [PATCH 2/2] [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/ 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. Generated-by: Claude Opus 4.8 (1M context) --- .../task-manager-metrics.component.html | 6 +- .../task-manager-metrics.component.spec.ts | 74 +++++++++++++++++ .../metrics/task-manager-metrics.component.ts | 20 +++-- .../status/task-manager-status.component.html | 77 ++++++++++-------- .../task-manager-status.component.spec.ts | 79 +++++++++++++++++++ .../status/task-manager-status.component.ts | 27 ++++--- .../task-manager-detail-poll.spec.ts | 77 ++++++++++++++++++ .../task-manager/task-manager-detail-poll.ts | 58 ++++++++++++++ .../src/app/services/task-manager.service.ts | 7 +- 9 files changed, 360 insertions(+), 65 deletions(-) create mode 100644 flink-runtime-web/web-dashboard/src/app/pages/task-manager/metrics/task-manager-metrics.component.spec.ts create mode 100644 flink-runtime-web/web-dashboard/src/app/pages/task-manager/status/task-manager-status.component.spec.ts create mode 100644 flink-runtime-web/web-dashboard/src/app/pages/task-manager/task-manager-detail-poll.spec.ts create mode 100644 flink-runtime-web/web-dashboard/src/app/pages/task-manager/task-manager-detail-poll.ts 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 07f735be986ff7..e66d71deb8eeea 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. --> - + - +
- + { + let fixture: ComponentFixture; + let element: HTMLElement; + const loadManager = vi.fn(); + const loadMetrics = vi.fn(); + + beforeEach(async () => { + loadManager.mockReset().mockReturnValue(of(mockDetail)); + loadMetrics.mockReset().mockReturnValue(of({ 'Status.JVM.Memory.Heap.Used': 1, 'Status.JVM.Memory.Heap.Max': 2 })); + await TestBed.configureTestingModule({ + imports: [TaskManagerMetricsComponent], + providers: [ + { provide: StatusService, useValue: { refresh$: of(true) } }, + { provide: TaskManagerService, useValue: { loadManager, loadMetrics } }, + { provide: ActivatedRoute, useValue: { parent: { snapshot: { params: { taskManagerId: 'tm-container-7' } } } } } + ] + }).compileComponents(); + fixture = TestBed.createComponent(TaskManagerMetricsComponent); + element = fixture.nativeElement as HTMLElement; + }); + + it('loads the metrics when the TaskManager is available', () => { + // Drive the load path without a full DOM render: the metric cards bind + // taskManagerDetail.metrics.*, so rendering needs a complete metrics fixture + // (the production build covers that). Here we assert the wiring. + fixture.componentInstance.ngOnInit(); + + expect(fixture.componentInstance.taskManagerDetail).toBe(mockDetail); + expect(loadMetrics).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(element.textContent).not.toContain('Flink Memory Model'); + expect(element.textContent).not.toContain('Advanced'); + expect(element.textContent).not.toContain('NaN'); + expect(loadMetrics).not.toHaveBeenCalled(); + }); +}); 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 c7c4350aa239e8..a3498f68cafb15 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 46e954d7082e92..8c4be8f6f9379f 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 @@ --> -
- {{ taskManagerDetail?.id }} - -
- - - {{ taskManagerDetail.path }} - - - {{ taskManagerDetail.freeSlots }} / {{ taskManagerDetail.slotsNumber }} - - - {{ taskManagerDetail.assignedTasks }} - - - {{ taskManagerDetail.timeSinceLastHeartbeat | date: 'yyyy-MM-dd HH:mm:ss' }} - - - {{ taskManagerDetail.dataPort }} - - - {{ taskManagerDetail.hardware.cpuCores }} - - - {{ taskManagerDetail.hardware.physicalMemory | humanizeBytes }} - - - {{ taskManagerDetail.hardware.freeMemory | humanizeBytes }} - - - {{ taskManagerDetail.hardware.managedMemory | humanizeBytes }} - - - + + +
+ {{ taskManagerDetail?.id }} + +
+ + + {{ taskManagerDetail.path }} + + + {{ taskManagerDetail.freeSlots }} / {{ taskManagerDetail.slotsNumber }} + + + {{ taskManagerDetail.assignedTasks }} + + + {{ taskManagerDetail.timeSinceLastHeartbeat | date: 'yyyy-MM-dd HH:mm:ss' }} + + + {{ taskManagerDetail.dataPort }} + + + {{ taskManagerDetail.hardware.cpuCores }} + + + {{ taskManagerDetail.hardware.physicalMemory | humanizeBytes }} + + + {{ taskManagerDetail.hardware.freeMemory | humanizeBytes }} + + + {{ taskManagerDetail.hardware.managedMemory | humanizeBytes }} + + + + 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 00000000000000..1f49dbb0ae4256 --- /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://flink@10.0.0.7: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; + 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 f579f7e518e71a..b3c2569cf09602 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(); @@ -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 00000000000000..b9947befd0d96f --- /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(); + 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(); + 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(); + 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 00000000000000..7635e8e0c26d1f --- /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, + load: () => Observable +): Observable { + 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/task-manager.service.ts b/flink-runtime-web/web-dashboard/src/app/services/task-manager.service.ts index 3a8c3227ddcf23..38d2837f09fa01 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 @@ -18,7 +18,7 @@ import { HttpClient, 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 { @@ -49,9 +49,8 @@ export class TaskManagerService { } loadManager(taskManagerId: string): Observable { - return this.httpClient - .get(`${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. + return this.httpClient.get(`${this.configService.BASE_URL}/taskmanagers/${taskManagerId}`); } loadLogList(taskManagerId: string): Observable {