diff --git a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
index b7ab854eb98a8..98c02d7a418fc 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 ada573ef5df9c..24ecb76ed2e96 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 f282db00ea979..890ac91ef3ce0 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 6dc04cd6d88d3..3fb29687d560a 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.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.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 ApplicationExceptionsHandler(
             MessageHeaders<
                             EmptyRequestBody,
                             ApplicationExceptionsInfoWithHistory,
-                            ApplicationMessageParameters>
+                            ApplicationExceptionsMessageParameters>
                     messageHeaders) {
         super(leaderRetriever, timeout, responseHeaders, messageHeaders);
     }
@@ -66,13 +70,20 @@ public 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 Collection<ArchivedJson> archiveApplicationWithPath(
                 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 aa4d36425d8e1..42a30a6eb63bf 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
@@ -91,10 +91,11 @@ public String toString() {
     }
 
     public static ApplicationExceptionsInfoWithHistory fromApplicationExceptionHistory(
-            Collection<ApplicationExceptionHistoryEntry> exceptions) {
+            Collection<ApplicationExceptionHistoryEntry> exceptions, int maxSize) {
         return new ApplicationExceptionsInfoWithHistory(
                 new ApplicationExceptionHistory(
                         exceptions.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 22a0ad90e0315..0f9c1b6c52315 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.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 HttpResponseStatus getResponseStatusCode() {
     }
 
     @Override
-    public ApplicationMessageParameters getUnresolvedMessageParameters() {
-        return new ApplicationMessageParameters();
+    public ApplicationExceptionsMessageParameters getUnresolvedMessageParameters() {
+        return new ApplicationExceptionsMessageParameters();
     }
 
     @Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/application/ApplicationExceptionsMessageParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/application/ApplicationExceptionsMessageParameters.java
index cee07b8b20966..62db2c9351c30 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/application/ApplicationExceptionsMessageParameters.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/application/ApplicationExceptionsMessageParameters.java
@@ -30,7 +30,7 @@
 /** {@link MessageParameters} for {@link ApplicationExceptionsHandler}. */
 public class ApplicationExceptionsMessageParameters extends ApplicationMessageParameters {
 
-    private final UpperLimitExceptionParameter upperLimitExceptionParameter =
+    public final UpperLimitExceptionParameter upperLimitExceptionParameter =
             new UpperLimitExceptionParameter();
 
     @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 26c11a60409de..68520eb537112 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.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,116 @@ void testExceptionWithJobId() throws Exception {
         assertThat(exceptionInfo.getJobId()).isEqualTo(jobId);
     }
 
+    @Test
+    void testMaxExceptionsLimitsHistorySize() throws Exception {
+        final List<ApplicationExceptionHistoryEntry> exceptionHistory = new ArrayList<>();
+        for (int i = 0; i < 5; 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> limitedRequest =
+                createRequest(archivedApplication.getApplicationId(), 2);
+
+        final ApplicationExceptionsInfoWithHistory response =
+                handler.handleRequest(limitedRequest, testingRestfulGateway).get();
+
+        assertThat(response.getExceptionHistory().getEntries()).hasSize(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 List<ApplicationExceptionHistoryEntry> exceptionHistory = new ArrayList<>();
+        for (int i = 0; i < 3; 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(), 10);
+
+        final ApplicationExceptionsInfoWithHistory response =
+                handler.handleRequest(oversizedRequest, testingRestfulGateway).get();
+
+        assertThat(response.getExceptionHistory().getEntries()).hasSize(3);
+    }
+
     @Test
     void testExceptionWithoutJobId() throws Exception {
         final RuntimeException rootCause = new RuntimeException("exception #0");