[FLINK-21867][webui] Expose concurrent exceptions

flink-runtime-web/web-dashboard/src/app/interfaces/job-exception.ts

@@ -35,7 +35,7 @@ export interface JobExceptionItemInterface {
 }
 
 export interface JobExceptionHistoryInterface {
- entries: ExceptionInfoInterface[];
+ entries: RootExceptionInfoInterface[];
  truncated: boolean;
 }
 
@@ -46,3 +46,7 @@ export interface ExceptionInfoInterface {
  taskName: string;
  location: string;
 }
+
+export interface RootExceptionInfoInterface extends ExceptionInfoInterface {
+ concurrentExceptions: ExceptionInfoInterface[];
+}

flink-runtime-web/web-dashboard/src/app/pages/job/exceptions/job-exceptions.component.html

@@ -24,39 +24,41 @@
  <nz-table
  class="no-border small"
  [nzSize]="'small'"
- [nzData]="listOfException"
+ [nzData]="exceptionHistory"
  [nzShowPagination]="false"
  [nzFrontPagination]="false">
  <thead>
  <tr>
  <th nzWidth="60px"></th>
  <th>Time</th>
  <th>Exception</th>
- <th>Name</th>
+ <th>Task Name</th>
  <th>Location</th>
  </tr>
  </thead>
  <tbody>
- <ng-container *ngFor="let exception of listOfException;trackBy:trackExceptionBy;">
+ <ng-container *ngFor="let item of exceptionHistory;trackBy:trackExceptionBy;">
  <tr>
- <td nzShowExpand [(nzExpand)]="exception.expand"></td>
- <td>{{exception.timestamp | date:'yyyy-MM-dd HH:mm:ss'}}</td>
- <td><div class="name">{{exception.exceptionName}}</div></td>
- <td>
- <div class="name">
- {{exception.taskName || "(global failure)"}}
- </div>
+ <td nzShowExpand [(nzExpand)]="item.expand"></td>
+ <td>{{item.selected.timestamp | date:'yyyy-MM-dd HH:mm:ss'}}</td>
+ <td><div class="name">{{item.selected.exceptionName}}</div></td>
+ <td class="select-td">
+ <nz-select class="exception-select" [nzDisabled]="item.exceptions.length == 1" nzShowSearch [(ngModel)]="item.selected">
+ <nz-option *ngFor="let ex of item.exceptions;" nzCustomContent [nzLabel]="ex.taskName" [nzValue]="ex">
+ <span nz-tooltip [nzTitle]="ex.taskName" nzMouseEnterDelay="0.5" nzPlacement="left">{{ ex.taskName }}</span>
+ </nz-option>
+ </nz-select>
  </td>
- <td>{{exception.location || "(unassigned)"}}</td>
+ <td>{{item.selected.location || "(unassigned)"}}</td>
  </tr>
- <tr [nzExpand]="exception.expand">
+ <tr [nzExpand]="item.expand">
  <td colspan="5" class="expand-td">
- <flink-monaco-editor *ngIf="exception.expand" class="subtask" [value]="exception.stacktrace"></flink-monaco-editor>
+ <flink-monaco-editor *ngIf="item.expand" class="subtask" [value]="item.selected.stacktrace"></flink-monaco-editor>
  </td>
  </tr>
  </ng-container>
- <tr *ngIf="listOfException.length > 0">
- <td colspan="6">
+ <tr *ngIf="exceptionHistory.length > 0">
+ <td colspan="5">
  <i nz-icon nzType="info-circle" nzTheme="fill"></i>&nbsp;
  <i>The exception history is limited to the most recent failures that caused parts of the job or the entire job to restart. The maximum history size can be configured via the Flink configuration property <b>web.exception-history-size</b>.</i>
  </td>

flink-runtime-web/web-dashboard/src/app/pages/job/exceptions/job-exceptions.component.less

@@ -40,3 +40,7 @@ flink-monaco-editor {
 .expand-td {
  padding: 0 !important;
 }
+
+.exception-select {
+ width: 300px;
+}

flink-runtime-web/web-dashboard/src/app/pages/job/exceptions/job-exceptions.component.ts

@@ -18,10 +18,40 @@
 
 import { formatDate } from '@angular/common';
 import { Component, OnInit, ChangeDetectionStrategy, ChangeDetectorRef } from '@angular/core';
-import { ExceptionInfoInterface } from 'interfaces';
+import { ExceptionInfoInterface, RootExceptionInfoInterface } from 'interfaces';
 import { distinctUntilChanged, flatMap, tap } from 'rxjs/operators';
 import { JobService } from 'services';
 
+interface ExceptionHistoryItem {
+
+ /**
+ * List of concurrent exceptions that caused this failure.
+ */
+ exceptions: ExceptionInfoInterface[];
+
+ /**
+ * An exception from the list, that is currently selected for rendering.
+ */
+ selected: ExceptionInfoInterface;
+
+ /**
+ * Should this failure be expanded in UI?
+ */
+ expand: boolean;
+}
+
+const stripConcurrentExceptions = function (rootException: RootExceptionInfoInterface): ExceptionInfoInterface {
+ const {concurrentExceptions, ...mainException} = rootException;
+ return mainException;
+}
+
+const markGlobalFailure = function (exception: ExceptionInfoInterface) {
+ if (exception.taskName == null) {
+ exception.taskName = '(global failure)';
+ }
+ return exception;
+}
+
 @Component({
  selector: 'flink-job-exceptions',
  templateUrl: './job-exceptions.component.html',
@@ -30,7 +60,7 @@ import { JobService } from 'services';
 })
 export class JobExceptionsComponent implements OnInit {
  rootException = '';
- listOfException: ExceptionInfoInterface[] = [];
+ exceptionHistory: ExceptionHistoryItem[] = [];
  truncated = false;
  isLoading = false;
  maxExceptions = 0;
@@ -53,15 +83,22 @@ export class JobExceptionsComponent implements OnInit {
  )
  .subscribe(data => {
  // @ts-ignore
- var exceptionHistory = data.exceptionHistory
+ const exceptionHistory = data.exceptionHistory
  if (exceptionHistory.entries.length > 0) {
- var mostRecentException = exceptionHistory.entries[0]
+ const mostRecentException = exceptionHistory.entries[0]
  this.rootException = formatDate(mostRecentException.timestamp, 'yyyy-MM-dd HH:mm:ss', 'en') + '\n' + mostRecentException.stacktrace;
  } else {
  this.rootException = 'No Root Exception';
  }
  this.truncated = exceptionHistory.truncated;
- this.listOfException = exceptionHistory.entries;
+ this.exceptionHistory = exceptionHistory.entries.map(entry => {
+ const values = [stripConcurrentExceptions(entry)].concat(entry.concurrentExceptions).map(markGlobalFailure);
+ return {
+ selected: values[0],
+ exceptions: values,
+ expand: false
+ };
+ });
  });
  }