Copilot commented on code in PR #28060: URL: https://github.com/apache/flink/pull/28060#discussion_r3735206534
########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.ts: ########## @@ -0,0 +1,764 @@ +/* + * 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 { DatePipe, NgIf } from '@angular/common'; +import { + AfterViewInit, + ChangeDetectionStrategy, + ChangeDetectorRef, + Component, + ElementRef, + Input, + OnChanges, + OnDestroy, + SimpleChanges, + ViewChild +} from '@angular/core'; +import { BehaviorSubject, defer, EMPTY, of, Subject, timer } from 'rxjs'; +import { catchError, distinctUntilChanged, map, repeat, switchMap, takeUntil } from 'rxjs/operators'; + +import * as G2 from '@antv/g2'; +import { Chart } from '@antv/g2'; +import { HumanizeBytesPipe } from '@flink-runtime-web/components/humanize-bytes.pipe'; +import { HumanizeDurationPipe } from '@flink-runtime-web/components/humanize-duration.pipe'; +import { + CheckpointConfig, + CheckpointHistory, + CheckpointSubTask, + CompletedSubTaskCheckpointStatistics, + SubTaskCheckpointStatisticsItem, + VerticesItem +} from '@flink-runtime-web/interfaces'; +import { JobService } from '@flink-runtime-web/services'; +import { NzAlertModule } from 'ng-zorro-antd/alert'; +import { NzEmptyModule } from 'ng-zorro-antd/empty'; +import { NzSpinModule } from 'ng-zorro-antd/spin'; +import { NzToolTipModule } from 'ng-zorro-antd/tooltip'; + +type PhaseKey = 'start_delay' | 'alignment' | 'sync' | 'async'; + +interface StripDatum { + id: number; + idLabel: string; + duration: number; + status: 'COMPLETED' | 'SAVEPOINT' | 'IN_PROGRESS' | 'FAILED'; + rawStatus: string; + isSavepoint: boolean; + triggered: number; + acked: string; + size: number; +} + +interface TimelineDatum { + row: string; + operatorFull: string; + subtask: number; + phase: PhaseKey; + phaseLabel: string; + range: [number, number]; + durationMs: number; + unaligned: boolean; + outlier: boolean; + aborted: boolean; + stateSize: number; + checkpointedSize: number; + totalDuration: number; +} + +const STRIP_COLOR: Record<StripDatum['status'], string> = { + COMPLETED: '#52c41a', + SAVEPOINT: '#722ed1', + IN_PROGRESS: '#faad14', + FAILED: '#f5222d' +}; + +const PHASE_COLOR: Record<PhaseKey, string> = { + start_delay: '#b8c0cc', + alignment: '#91caff', + sync: '#fa8c16', + async: '#52c41a' +}; + +const PHASE_LABEL: Record<PhaseKey, string> = { + start_delay: 'start_delay', + alignment: 'alignment', + sync: 'sync', + async: 'async' +}; + +const OPERATOR_NAME_MAX = 60; + +// Local HH:MM:SS.mmm — matches the header's DatePipe default (browser timezone). +function formatWallClock(epochMs: number): string { + const d = new Date(epochMs); + const pad = (n: number, len = 2): string => String(n).padStart(len, '0'); + return `${pad(d.getHours())}:${pad(d.getMinutes())}:${pad(d.getSeconds())}.${pad(d.getMilliseconds(), 3)}`; +} + +function buildTimelineTicks(min: number, max: number, target = 5): number[] { + const span = Math.max(max - min, 1); + const niceSteps = [1, 2, 5, 10, 20, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 30000, 60000]; + const desired = span / target; + const step = niceSteps.find(s => s >= desired) ?? span; + const ticks: number[] = []; + for (let t = min; t <= max + 0.5; t += step) { + ticks.push(t); + } + if (ticks[ticks.length - 1] < max) { + ticks.push(max); + } + return ticks; +} + +@Component({ + selector: 'flink-job-checkpoints-timeline', + templateUrl: './job-checkpoints-timeline.component.html', + styleUrls: ['./job-checkpoints-timeline.component.less'], + changeDetection: ChangeDetectionStrategy.OnPush, + imports: [ + NgIf, + DatePipe, + HumanizeDurationPipe, + HumanizeBytesPipe, + NzAlertModule, + NzEmptyModule, + NzSpinModule, + NzToolTipModule + ] +}) +export class JobCheckpointsTimelineComponent implements AfterViewInit, OnChanges, OnDestroy { + @Input() public jobId: string; + @Input() public vertices: VerticesItem[] = []; + @Input() public history: CheckpointHistory[] = []; + @Input() public config?: CheckpointConfig; + + @ViewChild('stripContainer', { static: true }) private readonly stripContainer: ElementRef<HTMLDivElement>; + @ViewChild('timelineContainer', { static: true }) private readonly timelineContainer: ElementRef<HTMLDivElement>; + + public selected?: StripDatum; + public loadingTimeline = false; + public timelineEmpty = false; + public loadFailed = false; + public failedVertexCount = 0; + public requestedVertexCount = 0; + + // Diverges from @Input once polling starts feeding live updates. + private currentHistory: CheckpointHistory[] = []; + public userPinned = false; + private stripChart?: Chart; + private timelineChart?: Chart; + private resizeObserver?: ResizeObserver; + private readonly destroy$ = new Subject<void>(); + private readonly loadTimeline$ = new Subject<StripDatum>(); + // Completed-checkpoint stats are immutable; memoize by id for instant re-selects. + private readonly detailsCache = new Map<number, Record<string, CheckpointSubTask | null>>(); + private readonly visible$ = new BehaviorSubject<boolean>(false); + private static readonly POLL_MIN_MS = 1500; + private static readonly POLL_MAX_MS = 15000; + private static readonly STRIP_MAX_BARS = 60; + private static readonly DETAILS_CACHE_MAX = 100; + // Flink sends Long.MAX_VALUE (well above MAX_SAFE_INTEGER) when periodic checkpointing is disabled. + private static readonly DISABLED_INTERVAL_THRESHOLD = Number.MAX_SAFE_INTEGER; + private static readonly TIMELINE_ROW_HEIGHT = 22; + private static readonly TIMELINE_HEIGHT_BASE = 80; + private static readonly TIMELINE_HEIGHT_MIN = 160; + private static readonly LABEL_CHAR_WIDTH = 7; + private static readonly LABEL_PADDING_BUFFER = 24; + private static readonly LABEL_PADDING_MIN = 260; + private static readonly LABEL_PADDING_MAX = 520; + + private readonly humanizeBytesPipe = new HumanizeBytesPipe(); + private readonly humanizeDurationPipe = new HumanizeDurationPipe(); + + constructor(private readonly jobService: JobService, private readonly cdr: ChangeDetectorRef) {} + + public ngAfterViewInit(): void { + this.currentHistory = this.history || []; + this.renderStrip(); + this.subscribeTimelineLoads(); + if (this.currentHistory.length > 0) { + this.selectInternal(this.toStripDatum(this.currentHistory[0]), false); + } + this.startPolling(); + // G2 locks autoFit at construction; re-fit + resume polling once width is real. + this.visible$.next(this.stripContainer.nativeElement.clientWidth > 0); + if (typeof ResizeObserver !== 'undefined') { + this.resizeObserver = new ResizeObserver(() => { + const stripVisible = this.stripContainer.nativeElement.clientWidth > 0; + if (this.stripChart && stripVisible) { + this.stripChart.forceFit(); + } + if (this.timelineChart && this.timelineContainer.nativeElement.clientWidth > 0) { + this.timelineChart.forceFit(); + } + this.visible$.next(stripVisible); + }); + this.resizeObserver.observe(this.stripContainer.nativeElement); + this.resizeObserver.observe(this.timelineContainer.nativeElement); + } + } + + public ngOnChanges(changes: SimpleChanges): void { + if (!this.stripContainer || !this.stripContainer.nativeElement) { + return; + } + if (changes['jobId'] && !changes['jobId'].firstChange) { + this.userPinned = false; + this.selected = undefined; + this.currentHistory = []; + this.clearTimeline(); + this.renderStrip(); + } Review Comment: `detailsCache` is keyed only by checkpoint id and is not cleared when the `jobId` input changes. If the user navigates to another job (same component instance) and checkpoint ids overlap (common starting at 1), cached per-vertex subtask details from the previous job can be shown for the new job. This issue also appears on line 502 of the same file. ########## flink-runtime-web/web-dashboard/src/app/services/job.service.ts: ########## @@ -183,6 +183,23 @@ export class JobService { ); } + public loadCheckpointAllSubtaskDetails( + jobId: string, + checkPointId: number, + vertexIds: string[] + ): Observable<Record<string, CheckpointSubTask | null>> { + if (vertexIds.length === 0) { + return of({} as Record<string, CheckpointSubTask | null>); + } + const requests = vertexIds.map(vid => + this.loadCheckpointSubtaskDetails(jobId, checkPointId, vid).pipe( + catchError(() => of<CheckpointSubTask | null>(null)), + map(result => ({ [vid]: result })) + ) + ); + return forkJoin(requests).pipe(map(parts => Object.assign({}, ...parts))); + } Review Comment: `loadCheckpointAllSubtaskDetails` fires one HTTP request per vertex in a single `forkJoin`, which can create very high parallelism (and repeated bursts if the user follows newest). For large jobs this can overwhelm the browser connection pool and the JobManager, and it also makes the whole call wait for the slowest request. Consider chunking the requests (bounded concurrency) while still returning a single merged map. ########## flink-runtime-web/web-dashboard/src/app/app.interceptor.ts: ########## @@ -59,9 +59,18 @@ export class AppInterceptor implements HttpInterceptor { window.location.href = String(res.headers.get('Location')); } + // A per-subtask checkpoint fetch can race the JM reaping that checkpoint (or hit it + // before it acks), yielding a benign 404. Suppress only that case; real 5xx/auth on + // the same path still surface. + const isExpectedSubtaskNotFound = + res instanceof HttpResponseBase && + res.status === HttpStatusCode.NotFound && + /\/checkpoints\/details\/\d+\/subtasks\//.test(res.url || ''); + Review Comment: This interceptor suppresses the global error toast for *all* 404s on `/checkpoints/details/<id>/subtasks/...`, regardless of whether the checkpoint is merely "in progress" (race-before-ack) vs genuinely unavailable (e.g., retained history expired, wrong vertex id). This is broader than the PR description (“while a checkpoint is still in progress”) and also affects existing pages like the subtask checkpoint view that rely on the toast to explain why the data didn’t load. Consider scoping the suppression to timeline-originated requests (e.g., set a custom request header in the timeline fetches and check it here), or otherwise gating it on an in-progress checkpoint context. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
