This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new d6b9ff65e29 [FLINK-39730][runtime] Honor `maxExceptions` on
`ApplicationExceptionsHandler`
d6b9ff65e29 is described below
commit d6b9ff65e29e1c7acbcfba450a5f3e005f0ec84f
Author: Jubin Soni <[email protected]>
AuthorDate: Fri Jun 19 03:41:36 2026 -0700
[FLINK-39730][runtime] Honor `maxExceptions` on
`ApplicationExceptionsHandler`
---
.../shortcodes/generated/rest_v1_dispatcher.html | 10 ++
docs/static/generated/rest_v1_dispatcher.yml | 9 ++
.../src/test/resources/rest_api_v1.snapshot | 5 +-
.../application/ApplicationExceptionsHandler.java | 22 +++-
.../ApplicationExceptionsInfoWithHistory.java | 11 +-
.../application/ApplicationExceptionsHeaders.java | 7 +-
.../ApplicationExceptionsHandlerTest.java | 141 ++++++++++++++++++++-
7 files changed, 190 insertions(+), 15 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
index b7ab854eb98..98c02d7a418 100644
--- a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
+++ b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
@@ -251,6 +251,16 @@
</ul>
</td>
</tr>
+ <tr>
+ <td colspan="2">Query parameters</td>
+ </tr>
+ <tr>
+ <td colspan="2">
+ <ul>
+<li><code>maxExceptions</code> (optional): Comma-separated list of integer
values that specifies the upper limit of exceptions to return.</li>
+ </ul>
+ </td>
+ </tr>
<tr>
<td colspan="2">
<label>
diff --git a/docs/static/generated/rest_v1_dispatcher.yml
b/docs/static/generated/rest_v1_dispatcher.yml
index ada573ef5df..24ecb76ed2e 100644
--- a/docs/static/generated/rest_v1_dispatcher.yml
+++ b/docs/static/generated/rest_v1_dispatcher.yml
@@ -62,6 +62,15 @@ paths:
required: true
schema:
$ref: "#/components/schemas/ApplicationID"
+ - name: maxExceptions
+ in: query
+ description: Comma-separated list of integer values that specifies the
upper
+ limit of exceptions to return.
+ required: false
+ style: form
+ schema:
+ type: integer
+ format: int32
responses:
"200":
description: The request was successful.
diff --git a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
index f282db00ea9..890ac91ef3c 100644
--- a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
+++ b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
@@ -180,7 +180,10 @@
} ]
},
"query-parameters" : {
- "queryParameters" : [ ]
+ "queryParameters" : [ {
+ "key" : "maxExceptions",
+ "mandatory" : false
+ } ]
},
"request" : {
"type" : "object",
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/application/ApplicationExceptionsHandler.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/application/ApplicationExceptionsHandler.java
index 6dc04cd6d88..3fb29687d56 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/application/ApplicationExceptionsHandler.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/application/ApplicationExceptionsHandler.java
@@ -24,9 +24,10 @@ import
org.apache.flink.runtime.rest.handler.AbstractRestHandler;
import org.apache.flink.runtime.rest.handler.HandlerRequest;
import
org.apache.flink.runtime.rest.messages.ApplicationExceptionsInfoWithHistory;
import org.apache.flink.runtime.rest.messages.ApplicationIDPathParameter;
-import org.apache.flink.runtime.rest.messages.ApplicationMessageParameters;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import
org.apache.flink.runtime.rest.messages.application.ApplicationExceptionsMessageParameters;
+import org.apache.flink.runtime.rest.messages.job.UpperLimitExceptionParameter;
import org.apache.flink.runtime.webmonitor.RestfulGateway;
import org.apache.flink.runtime.webmonitor.history.ApplicationJsonArchivist;
import org.apache.flink.runtime.webmonitor.history.ArchivedJson;
@@ -38,6 +39,7 @@ import java.io.IOException;
import java.time.Duration;
import java.util.Collection;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
@@ -47,9 +49,11 @@ public class ApplicationExceptionsHandler
RestfulGateway,
EmptyRequestBody,
ApplicationExceptionsInfoWithHistory,
- ApplicationMessageParameters>
+ ApplicationExceptionsMessageParameters>
implements ApplicationJsonArchivist {
+ static final int MAX_NUMBER_EXCEPTION_TO_REPORT = 20;
+
public ApplicationExceptionsHandler(
GatewayRetriever<? extends RestfulGateway> leaderRetriever,
Duration timeout,
@@ -57,7 +61,7 @@ public class ApplicationExceptionsHandler
MessageHeaders<
EmptyRequestBody,
ApplicationExceptionsInfoWithHistory,
- ApplicationMessageParameters>
+ ApplicationExceptionsMessageParameters>
messageHeaders) {
super(leaderRetriever, timeout, responseHeaders, messageHeaders);
}
@@ -66,13 +70,20 @@ public class ApplicationExceptionsHandler
public CompletableFuture<ApplicationExceptionsInfoWithHistory>
handleRequest(
@Nonnull HandlerRequest<EmptyRequestBody> request, @Nonnull
RestfulGateway gateway) {
ApplicationID applicationId =
request.getPathParameter(ApplicationIDPathParameter.class);
+ final List<Integer> exceptionToReportMaxSizes =
+ request.getQueryParameter(UpperLimitExceptionParameter.class);
+ final int exceptionToReportMaxSize =
+ exceptionToReportMaxSizes.size() > 0
+ ? exceptionToReportMaxSizes.get(0)
+ : MAX_NUMBER_EXCEPTION_TO_REPORT;
return gateway.requestApplication(applicationId, timeout)
.thenApply(
archivedApplication ->
ApplicationExceptionsInfoWithHistory
.fromApplicationExceptionHistory(
-
archivedApplication.getExceptionHistory()));
+
archivedApplication.getExceptionHistory(),
+ exceptionToReportMaxSize));
}
@Override
@@ -88,6 +99,7 @@ public class ApplicationExceptionsHandler
new ArchivedJson(
path,
ApplicationExceptionsInfoWithHistory.fromApplicationExceptionHistory(
- archivedApplication.getExceptionHistory())));
+ archivedApplication.getExceptionHistory(),
+ MAX_NUMBER_EXCEPTION_TO_REPORT)));
}
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ApplicationExceptionsInfoWithHistory.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ApplicationExceptionsInfoWithHistory.java
index aa4d36425d8..82f97c2013e 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ApplicationExceptionsInfoWithHistory.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ApplicationExceptionsInfoWithHistory.java
@@ -33,7 +33,9 @@ import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotatio
import javax.annotation.Nullable;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.StringJoiner;
@@ -91,10 +93,15 @@ public class ApplicationExceptionsInfoWithHistory
implements ResponseBody {
}
public static ApplicationExceptionsInfoWithHistory
fromApplicationExceptionHistory(
- Collection<ApplicationExceptionHistoryEntry> exceptions) {
+ Collection<ApplicationExceptionHistoryEntry> exceptions, int
maxSize) {
+ // reverse so the newest entries are returned first when truncating by
maxSize,
+ // matching JobExceptionsHandler#createJobExceptionHistory semantics
+ final List<ApplicationExceptionHistoryEntry> reversed = new
ArrayList<>(exceptions);
+ Collections.reverse(reversed);
return new ApplicationExceptionsInfoWithHistory(
new ApplicationExceptionHistory(
- exceptions.stream()
+ reversed.stream()
+ .limit(maxSize)
.map(
exception ->
new ApplicationExceptionInfo(
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/application/ApplicationExceptionsHeaders.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/application/ApplicationExceptionsHeaders.java
index 22a0ad90e03..0f9c1b6c523 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/application/ApplicationExceptionsHeaders.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/application/ApplicationExceptionsHeaders.java
@@ -22,7 +22,6 @@ import org.apache.flink.runtime.rest.HttpMethodWrapper;
import
org.apache.flink.runtime.rest.handler.application.ApplicationExceptionsHandler;
import
org.apache.flink.runtime.rest.messages.ApplicationExceptionsInfoWithHistory;
import org.apache.flink.runtime.rest.messages.ApplicationIDPathParameter;
-import org.apache.flink.runtime.rest.messages.ApplicationMessageParameters;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
import org.apache.flink.runtime.rest.messages.RuntimeMessageHeaders;
@@ -33,7 +32,7 @@ public class ApplicationExceptionsHeaders
implements RuntimeMessageHeaders<
EmptyRequestBody,
ApplicationExceptionsInfoWithHistory,
- ApplicationMessageParameters> {
+ ApplicationExceptionsMessageParameters> {
private static final ApplicationExceptionsHeaders INSTANCE = new
ApplicationExceptionsHeaders();
@@ -58,8 +57,8 @@ public class ApplicationExceptionsHeaders
}
@Override
- public ApplicationMessageParameters getUnresolvedMessageParameters() {
- return new ApplicationMessageParameters();
+ public ApplicationExceptionsMessageParameters
getUnresolvedMessageParameters() {
+ return new ApplicationExceptionsMessageParameters();
}
@Override
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/application/ApplicationExceptionsHandlerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/application/ApplicationExceptionsHandlerTest.java
index 26c11a60409..9f26e84d1f6 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/application/ApplicationExceptionsHandlerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/application/ApplicationExceptionsHandlerTest.java
@@ -27,9 +27,10 @@ import org.apache.flink.runtime.rest.handler.HandlerRequest;
import org.apache.flink.runtime.rest.handler.HandlerRequestException;
import
org.apache.flink.runtime.rest.messages.ApplicationExceptionsInfoWithHistory;
import org.apache.flink.runtime.rest.messages.ApplicationIDPathParameter;
-import org.apache.flink.runtime.rest.messages.ApplicationMessageParameters;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
import
org.apache.flink.runtime.rest.messages.application.ApplicationExceptionsHeaders;
+import
org.apache.flink.runtime.rest.messages.application.ApplicationExceptionsMessageParameters;
+import org.apache.flink.runtime.rest.messages.job.UpperLimitExceptionParameter;
import org.apache.flink.runtime.webmonitor.RestfulGateway;
import org.apache.flink.runtime.webmonitor.TestingRestfulGateway;
import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
@@ -57,13 +58,28 @@ class ApplicationExceptionsHandlerTest {
private static HandlerRequest<EmptyRequestBody>
createRequest(ApplicationID applicationId)
throws HandlerRequestException {
+ return createRequest(applicationId, Collections.emptyMap());
+ }
+
+ private static HandlerRequest<EmptyRequestBody> createRequest(
+ ApplicationID applicationId, int maxExceptions) throws
HandlerRequestException {
+ Map<String, List<String>> queryParameters = new HashMap<>();
+ queryParameters.put(
+ UpperLimitExceptionParameter.KEY,
+ Collections.singletonList(Integer.toString(maxExceptions)));
+ return createRequest(applicationId, queryParameters);
+ }
+
+ private static HandlerRequest<EmptyRequestBody> createRequest(
+ ApplicationID applicationId, Map<String, List<String>>
queryParameters)
+ throws HandlerRequestException {
Map<String, String> pathParameters = new HashMap<>();
pathParameters.put(ApplicationIDPathParameter.KEY,
applicationId.toString());
return HandlerRequest.resolveParametersAndCreate(
EmptyRequestBody.getInstance(),
- new ApplicationMessageParameters(),
+ new ApplicationExceptionsMessageParameters(),
pathParameters,
- Collections.emptyMap(),
+ queryParameters,
Collections.emptyList());
}
@@ -143,6 +159,125 @@ class ApplicationExceptionsHandlerTest {
assertThat(exceptionInfo.getJobId()).isEqualTo(jobId);
}
+ @Test
+ void testMaxExceptionsLimitsHistorySize() throws Exception {
+ final int historySize = 5;
+ final int maxExceptions = 2;
+ final long baseTimestamp = System.currentTimeMillis();
+ final List<ApplicationExceptionHistoryEntry> exceptionHistory = new
ArrayList<>();
+ for (int i = 0; i < historySize; i++) {
+ exceptionHistory.add(
+ new ApplicationExceptionHistoryEntry(
+ new RuntimeException("exception #" + i),
baseTimestamp + i, null));
+ }
+
+ final ArchivedApplication applicationWithExceptions =
+ new ArchivedApplication(
+ archivedApplication.getApplicationId(),
+ archivedApplication.getApplicationName(),
+ ApplicationState.FAILED,
+ new long[] {1L, 1L, 1L, 1L, 1L, 1L, 1L},
+ Collections.emptyMap(),
+ exceptionHistory);
+
+ testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setRequestApplicationFunction(
+ applicationId ->
+ CompletableFuture.completedFuture(
+ applicationWithExceptions))
+ .build();
+
+ final HandlerRequest<EmptyRequestBody> limitedRequest =
+ createRequest(archivedApplication.getApplicationId(),
maxExceptions);
+
+ final ApplicationExceptionsInfoWithHistory response =
+ handler.handleRequest(limitedRequest,
testingRestfulGateway).get();
+
+ // newest entries are returned first when truncating, so with
historySize=5 and
+ // maxExceptions=2 we expect entries #4 and #3 (in that order)
+ assertThat(response.getExceptionHistory().getEntries())
+ .hasSize(maxExceptions)
+ .extracting(
+
ApplicationExceptionsInfoWithHistory.ApplicationExceptionInfo::getTimestamp)
+ .containsExactly(baseTimestamp + historySize - 1,
baseTimestamp + historySize - 2);
+ }
+
+ @Test
+ void testDefaultCapAppliedWhenMaxExceptionsNotProvided() throws Exception {
+ final int totalExceptions =
ApplicationExceptionsHandler.MAX_NUMBER_EXCEPTION_TO_REPORT + 5;
+ final List<ApplicationExceptionHistoryEntry> exceptionHistory = new
ArrayList<>();
+ for (int i = 0; i < totalExceptions; i++) {
+ exceptionHistory.add(
+ new ApplicationExceptionHistoryEntry(
+ new RuntimeException("exception #" + i),
+ System.currentTimeMillis(),
+ null));
+ }
+
+ final ArchivedApplication applicationWithExceptions =
+ new ArchivedApplication(
+ archivedApplication.getApplicationId(),
+ archivedApplication.getApplicationName(),
+ ApplicationState.FAILED,
+ new long[] {1L, 1L, 1L, 1L, 1L, 1L, 1L},
+ Collections.emptyMap(),
+ exceptionHistory);
+
+ testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setRequestApplicationFunction(
+ applicationId ->
+ CompletableFuture.completedFuture(
+ applicationWithExceptions))
+ .build();
+
+ final ApplicationExceptionsInfoWithHistory response =
+ handler.handleRequest(handlerRequest,
testingRestfulGateway).get();
+
+ assertThat(response.getExceptionHistory().getEntries())
+
.hasSize(ApplicationExceptionsHandler.MAX_NUMBER_EXCEPTION_TO_REPORT);
+ }
+
+ @Test
+ void testMaxExceptionsLargerThanHistorySizeReturnsAllEntries() throws
Exception {
+ final int historySize = 3;
+ final int oversizedMaxExceptions = historySize + 7;
+ final List<ApplicationExceptionHistoryEntry> exceptionHistory = new
ArrayList<>();
+ for (int i = 0; i < historySize; i++) {
+ exceptionHistory.add(
+ new ApplicationExceptionHistoryEntry(
+ new RuntimeException("exception #" + i),
+ System.currentTimeMillis(),
+ null));
+ }
+
+ final ArchivedApplication applicationWithExceptions =
+ new ArchivedApplication(
+ archivedApplication.getApplicationId(),
+ archivedApplication.getApplicationName(),
+ ApplicationState.FAILED,
+ new long[] {1L, 1L, 1L, 1L, 1L, 1L, 1L},
+ Collections.emptyMap(),
+ exceptionHistory);
+
+ testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setRequestApplicationFunction(
+ applicationId ->
+ CompletableFuture.completedFuture(
+ applicationWithExceptions))
+ .build();
+
+ final HandlerRequest<EmptyRequestBody> oversizedRequest =
+ createRequest(archivedApplication.getApplicationId(),
oversizedMaxExceptions);
+
+ final ApplicationExceptionsInfoWithHistory response =
+ handler.handleRequest(oversizedRequest,
testingRestfulGateway).get();
+
+
assertThat(response.getExceptionHistory().getEntries()).hasSize(historySize);
+ }
+
@Test
void testExceptionWithoutJobId() throws Exception {
final RuntimeException rootCause = new RuntimeException("exception
#0");
