[FLINK-34922][rest] Support concurrent global failure

apache · Mar 28, 2024 · dc957bf · dc957bf
1 parent 83f82ab
commit dc957bf
Show file tree

Hide file tree

Showing 2 changed files with 63 additions and 1 deletion.
diff --git a/...runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobExceptionsHandler.java b/...runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobExceptionsHandler.java
@@ -249,6 +249,19 @@ private static JobExceptionsInfoWithHistory.RootExceptionInfo createRootExceptio
 
  private static JobExceptionsInfoWithHistory.ExceptionInfo createExceptionInfo(
  ExceptionHistoryEntry exceptionHistoryEntry) {
+
+ if (exceptionHistoryEntry.isGlobal()) {
+ return new JobExceptionsInfoWithHistory.ExceptionInfo(
+ exceptionHistoryEntry.getException().getOriginalErrorClassName(),
+ exceptionHistoryEntry.getExceptionAsString(),
+ exceptionHistoryEntry.getTimestamp(),
+ exceptionHistoryEntry.getFailureLabels(),
+ null,
+ null,
+ null,
+ null);
+ }
+
  assertLocalExceptionInfo(exceptionHistoryEntry);
 
  return new JobExceptionsInfoWithHistory.ExceptionInfo(

diff --git a/...ime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobExceptionsHandlerTest.java b/...ime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobExceptionsHandlerTest.java
@@ -64,6 +64,7 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -216,6 +217,47 @@ void testWithExceptionHistory()
  assertThat(response.getExceptionHistory().isTruncated()).isFalse();
  }
 
+ @Test
+ void testWithExceptionHistoryAndConcurrentGlobalFailure()
+ throws HandlerRequestException, ExecutionException, InterruptedException {
+ final ExceptionHistoryEntry otherFailure =
+ ExceptionHistoryEntry.createGlobal(
+ new RuntimeException("exception #1"),
+ CompletableFuture.completedFuture(Collections.emptyMap()));
+ final RootExceptionHistoryEntry rootCause =
+ fromGlobalFailure(
+ new RuntimeException("exception #0"),
+ System.currentTimeMillis(),
+ Collections.singleton(otherFailure));
+
+ final ExecutionGraphInfo executionGraphInfo = createExecutionGraphInfo(rootCause);
+ final HandlerRequest<EmptyRequestBody> request =
+ createRequest(executionGraphInfo.getJobId(), 10);
+ final JobExceptionsInfoWithHistory response =
+ testInstance.handleRequest(request, executionGraphInfo);
+
+ assertThat(response.getExceptionHistory().getEntries())
+ .hasSize(1)
+ .satisfies(
+ matching(
+ contains(
+ historyContainsGlobalFailure(
+ rootCause.getException(),
+ rootCause.getTimestamp(),
+ matchesFailure(
+ otherFailure.getException(),
+ otherFailure.getTimestamp(),
+ otherFailure.getFailureLabelsFuture(),
+ otherFailure.getFailingTaskName(),
+ JobExceptionsHandler.toString(
+ otherFailure
+ .getTaskManagerLocation()),
+ JobExceptionsHandler.toTaskManagerId(
+ otherFailure
+ .getTaskManagerLocation()))))));
+ assertThat(response.getExceptionHistory().isTruncated()).isFalse();
+ }
+
  @Test
  void testWithExceptionHistoryWithMatchingFailureLabel()
  throws HandlerRequestException, ExecutionException, InterruptedException {
@@ -542,13 +584,20 @@ private static HandlerRequest<EmptyRequestBody> createRequest(
  }
 
  private static RootExceptionHistoryEntry fromGlobalFailure(Throwable cause, long timestamp) {
+ return fromGlobalFailure(cause, timestamp, Collections.emptySet());
+ }
+
+ private static RootExceptionHistoryEntry fromGlobalFailure(
+ Throwable cause,
+ long timestamp,
+ Collection<ExceptionHistoryEntry> concurrentExceptions) {
  return new RootExceptionHistoryEntry(
  cause,
  timestamp,
  FailureEnricherUtils.EMPTY_FAILURE_LABELS,
  null,
  null,
- Collections.emptySet());
+ concurrentExceptions);
  }
 
  // -------- factory methods for instantiating new Matchers --------