This is an automated email from the ASF dual-hosted git repository.
RocMarshal pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new e1125e34ce1 [FLINK-39490][web] Improve feedback when cluster is
unreachable (#27971)
e1125e34ce1 is described below
commit e1125e34ce13af21a85e88e24dc4ffb4cb9c4d67
Author: Purushottam Sinha <[email protected]>
AuthorDate: Fri Jul 31 21:45:43 2026 +0530
[FLINK-39490][web] Improve feedback when cluster is unreachable (#27971)
---
.../web-dashboard/src/app/app.component.ts | 4 +++-
.../web-dashboard/src/app/app.interceptor.ts | 27 +++++++++++++++++++++-
.../web-dashboard/src/app/services/job.service.ts | 7 +++---
.../src/app/services/status.service.ts | 19 ++++++++++++++-
4 files changed, 51 insertions(+), 6 deletions(-)
diff --git a/flink-runtime-web/web-dashboard/src/app/app.component.ts
b/flink-runtime-web/web-dashboard/src/app/app.component.ts
index 47fb9b66d47..daf4ada8c89 100644
--- a/flink-runtime-web/web-dashboard/src/app/app.component.ts
+++ b/flink-runtime-web/web-dashboard/src/app/app.component.ts
@@ -84,5 +84,7 @@ export class AppComponent {
constructor(
public statusService: StatusService,
private cdr: ChangeDetectorRef
- ) {}
+ ) {
+ this.statusService.registerAppCdr(this.cdr);
+ }
}
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 cfc85d20091..c77b890da33 100644
--- a/flink-runtime-web/web-dashboard/src/app/app.interceptor.ts
+++ b/flink-runtime-web/web-dashboard/src/app/app.interceptor.ts
@@ -21,12 +21,13 @@ import {
HttpHandler,
HttpInterceptor,
HttpRequest,
+ HttpResponse,
HttpResponseBase,
HttpStatusCode
} from '@angular/common/http';
import { Injectable } from '@angular/core';
import { Observable, throwError } from 'rxjs';
-import { catchError } from 'rxjs/operators';
+import { catchError, tap } from 'rxjs/operators';
import { StatusService } from '@flink-runtime-web/services';
import { NzNotificationService, NzNotificationDataOptions } from
'ng-zorro-antd/notification';
@@ -48,6 +49,16 @@ export class AppInterceptor implements HttpInterceptor {
};
return next.handle(req.clone({ withCredentials: true })).pipe(
+ tap(event => {
+ if (event instanceof HttpResponse) {
+ if (this.statusService.networkFailureCount > 0) {
+ this.statusService.networkFailureCount = 0;
+ }
+ if (this.statusService.networkErrorNotificationId) {
+
this.notificationService.remove(this.statusService.networkErrorNotificationId);
+ }
+ }
+ }),
catchError(res => {
if (
res instanceof HttpResponseBase &&
@@ -67,6 +78,20 @@ export class AppInterceptor implements HttpInterceptor {
) {
this.statusService.listOfErrorMessage.push(errorMessage);
this.notificationService.info('Server Response Message:',
errorMessage.replaceAll(' at ', '\n at '), option);
+ this.statusService.markAppForCheck();
+ } else if (res.status === 0 || res.status >= 500) {
+ this.statusService.networkFailureCount += 1;
+ if (
+ this.statusService.networkFailureCount >=
this.statusService.networkFailureThreshold &&
+ !this.statusService.networkErrorNotificationId
+ ) {
+ const ref = this.notificationService.warning('Network Error:',
'Connection lost or server error.', option);
+ this.statusService.networkErrorNotificationId = ref.messageId;
+ ref.onClose.subscribe(() => {
+ this.statusService.networkErrorNotificationId = null;
+ this.statusService.networkFailureCount = 0;
+ });
+ }
}
return throwError(res);
})
diff --git a/flink-runtime-web/web-dashboard/src/app/services/job.service.ts
b/flink-runtime-web/web-dashboard/src/app/services/job.service.ts
index 0250f051a67..2e2ca5d46ea 100644
--- a/flink-runtime-web/web-dashboard/src/app/services/job.service.ts
+++ b/flink-runtime-web/web-dashboard/src/app/services/job.service.ts
@@ -94,9 +94,10 @@ export class JobService {
}
public loadJob(jobId: string): Observable<JobDetailCorrect> {
- return this.httpClient
- .get<JobDetail>(`${this.configService.BASE_URL}/jobs/${jobId}`)
- .pipe(map(job => this.convertJob(job)));
+ return
this.httpClient.get<JobDetail>(`${this.configService.BASE_URL}/jobs/${jobId}`).pipe(
+ map(job => this.convertJob(job)),
+ catchError(() => EMPTY)
+ );
}
public loadAccumulators(jobId: string, vertexId: string):
Observable<JobAccumulators> {
diff --git a/flink-runtime-web/web-dashboard/src/app/services/status.service.ts
b/flink-runtime-web/web-dashboard/src/app/services/status.service.ts
index 3f82cfd4586..e932ba5ec4d 100644
--- a/flink-runtime-web/web-dashboard/src/app/services/status.service.ts
+++ b/flink-runtime-web/web-dashboard/src/app/services/status.service.ts
@@ -17,7 +17,7 @@
*/
import { HttpClient } from '@angular/common/http';
-import { inject, Injectable } from '@angular/core';
+import { ChangeDetectorRef, inject, Injectable } from '@angular/core';
import { NavigationEnd, Router } from '@angular/router';
import { EMPTY, fromEvent, interval, merge, Observable, Subject } from 'rxjs';
import { debounceTime, filter, map, share, startWith, switchMap, tap } from
'rxjs/operators';
@@ -37,6 +37,23 @@ export class StatusService {
/** Error server response message cache list. */
public listOfErrorMessage: string[] = [];
+ /** Threshold of consecutive network failures before surfacing an error to
the user. */
+ public readonly networkFailureThreshold = 5;
+ /** Count of consecutive network failures (status 0 or >=500) across all
requests. */
+ public networkFailureCount = 0;
+ /** messageId of the currently-visible network-error notification, if any. */
+ public networkErrorNotificationId: string | null = null;
+
+ private appCdr?: ChangeDetectorRef;
+
+ public registerAppCdr(cdr: ChangeDetectorRef): void {
+ this.appCdr = cdr;
+ }
+
+ public markAppForCheck(): void {
+ this.appCdr?.markForCheck();
+ }
+
/** Flink configuration from backend. */
public configuration: Configuration;
