diff --git a/flink-runtime-web/web-dashboard/src/app/interfaces/job-exception.ts b/flink-runtime-web/web-dashboard/src/app/interfaces/job-exception.ts
index 79a86c31e705d..16b5b8bc22904 100644
--- a/flink-runtime-web/web-dashboard/src/app/interfaces/job-exception.ts
+++ b/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[];
+}
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job/exceptions/job-exceptions.component.html b/flink-runtime-web/web-dashboard/src/app/pages/job/exceptions/job-exceptions.component.html
index 061adf47e08b1..2d3f36f728c84 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job/exceptions/job-exceptions.component.html
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job/exceptions/job-exceptions.component.html
@@ -24,7 +24,7 @@
     <nz-table
       class="no-border small"
       [nzSize]="'small'"
-      [nzData]="listOfException"
+      [nzData]="exceptionHistory"
       [nzShowPagination]="false"
       [nzFrontPagination]="false">
       <thead>
@@ -32,31 +32,33 @@
           <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>
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job/exceptions/job-exceptions.component.less b/flink-runtime-web/web-dashboard/src/app/pages/job/exceptions/job-exceptions.component.less
index 6865d3ad554a9..a44e8958ccfa2 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job/exceptions/job-exceptions.component.less
+++ b/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;
+}
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job/exceptions/job-exceptions.component.ts b/flink-runtime-web/web-dashboard/src/app/pages/job/exceptions/job-exceptions.component.ts
index 0513993390e8a..228d2690d9f73 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job/exceptions/job-exceptions.component.ts
+++ b/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
+          };
+        });
       });
   }