This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
commit fcb581f0039f9704b6eaf15a2fabaa4e05d79048
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 | 12 +++++
.../rest/handler/job/JobExceptionsHandlerTest.java | 53 +++++++++++++++++++++-
2 files changed, 64 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 5ece82a2671..84140c8c007 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
@@ -248,6 +248,18 @@ 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);
+ }
+
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 efce7903686..c7699c6f951 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
@@ -55,6 +55,7 @@ import org.apache.flink.util.SerializedThrowable;
import org.apache.flink.util.TestLogger;
import org.apache.flink.util.concurrent.Executors;
+import org.assertj.core.api.Assertions;
import org.hamcrest.Description;
import org.hamcrest.Matcher;
import org.hamcrest.TypeSafeDiagnosingMatcher;
@@ -64,6 +65,7 @@ import org.junit.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;
@@ -73,6 +75,7 @@ import java.util.concurrent.ExecutionException;
import java.util.function.Function;
import static
org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId;
+import static org.assertj.core.api.HamcrestCondition.matching;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.collection.IsEmptyCollection.empty;
@@ -214,6 +217,47 @@ public class JobExceptionsHandlerTest extends TestLogger {
assertFalse(response.getExceptionHistory().isTruncated());
}
+ @Test
+ public 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);
+
+ Assertions.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()))))));
+
Assertions.assertThat(response.getExceptionHistory().isTruncated()).isFalse();
+ }
+
@Test
public void testWithExceptionHistoryWithMatchingFailureLabel()
throws HandlerRequestException, ExecutionException,
InterruptedException {
@@ -532,13 +576,20 @@ public class JobExceptionsHandlerTest extends TestLogger {
}
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 --------
