This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit faa880c703cadba4521fc8ef885a242ded4b2ac7
Author: Chesnay Schepler <ches...@apache.org>
AuthorDate: Wed Mar 27 09:33:08 2024 +0100
[FLINK-34922][rest] Support concurrent global failure
---
.../rest/handler/job/JobExceptionsHandler.java | 13 ++++++
.../rest/handler/job/JobExceptionsHandlerTest.java | 51 +++++++++++++++++++++-
2 files changed, 63 insertions(+), 1 deletion(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobExceptionsHandler.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobExceptionsHandler.java
index 55c7875e85c..6d5f49d55b4 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobExceptionsHandler.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobExceptionsHandler.java
@@ -249,6 +249,19 @@ public class JobExceptionsHandler
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/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobExceptionsHandlerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobExceptionsHandlerTest.java
index f5354e5ce90..761881d1624 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobExceptionsHandlerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobExceptionsHandlerTest.java
@@ -63,6 +63,7 @@ import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -215,6 +216,47 @@ class JobExceptionsHandlerTest {
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 {
@@ -540,13 +582,20 @@ class JobExceptionsHandlerTest {
}
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 --------
