This is an automated email from the ASF dual-hosted git repository.
rmetzger 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 d5c4ba054b7 [FLINK-40530][runtime-web] Isolate per-entry failures in
job overview aggregation (#29062)
d5c4ba054b7 is described below
commit d5c4ba054b7e90465aa5229826385a67ad9c51c5
Author: Archit Goyal <[email protected]>
AuthorDate: Thu Sep 10 03:20:59 2026 -0700
[FLINK-40530][runtime-web] Isolate per-entry failures in job overview
aggregation (#29062)
A single malformed archive no longer aborts the whole /jobs/overview
update; it is skipped and logged, and a good overview is kept if all entries
fail.
Co-authored-by: argoyal2212 <[email protected]>
Generated-by: GitHub Copilot CLI
---
.../history/HistoryServerArchiveFetcher.java | 45 +++++++++++++++-----
.../history/HistoryServerArchiveFetcherTest.java | 48 ++++++++++++++++++++++
2 files changed, 82 insertions(+), 11 deletions(-)
diff --git
a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java
b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java
index 632551c8eb3..2c17b25cbeb 100644
---
a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java
+++
b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java
@@ -505,19 +505,42 @@ public class HistoryServerArchiveFetcher<Entry>
implements AutoCloseable {
try {
Collection<JobDetails> allJobs = new ArrayList<>();
List<Entry> overviews =
archiveStorage.getEntriesByPrefix(JOB_OVERVIEWS_KEY_PREFIX);
+ int skipped = 0;
for (Entry overview : overviews) {
- MultipleJobsDetails subJobs;
- // We treated File as a special case, mainly as a performance
trade-off to avoid the
- // overhead of loading the archive into string.
- if (overview instanceof File) {
- subJobs = mapper.readValue((File) overview,
MultipleJobsDetails.class);
- } else {
- subJobs =
- mapper.readValue(
-
archiveStorage.readArchiveContent(overview),
- MultipleJobsDetails.class);
+ try {
+ MultipleJobsDetails subJobs;
+ // We treated File as a special case, mainly as a
performance trade-off to
+ // avoid the overhead of loading the archive into string.
+ if (overview instanceof File) {
+ subJobs = mapper.readValue((File) overview,
MultipleJobsDetails.class);
+ } else {
+ subJobs =
+ mapper.readValue(
+
archiveStorage.readArchiveContent(overview),
+ MultipleJobsDetails.class);
+ }
+ allJobs.addAll(subJobs.getJobs());
+ } catch (Exception e) {
+ // A single malformed/incompatible archive (e.g. written
by a different Flink
+ // version) must not prevent the remaining archives from
being aggregated.
+ skipped++;
+ LOG.warn(
+ "Failed to parse job overview from entry {},
skipping it.",
+ overview,
+ e);
}
- allJobs.addAll(subJobs.getJobs());
+ }
+ if (skipped > 0) {
+ LOG.warn("Skipped {} job overview(s) that could not be
parsed.", skipped);
+ }
+ if (!overviews.isEmpty() && allJobs.isEmpty()) {
+ // Every entry failed to parse; keep the previously written
overview instead of
+ // replacing it with an empty one.
+ LOG.error(
+ "All {} job overview(s) failed to parse; keeping the
last known good "
+ + "combined overview instead of overwriting
it.",
+ overviews.size());
+ return;
}
String overviewWithJobs = mapper.writeValueAsString(new
MultipleJobsDetails(allJobs));
archiveStorage.putArchiveContent(
diff --git
a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java
b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java
index b95449337ec..877f419a588 100644
---
a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java
+++
b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java
@@ -346,6 +346,54 @@ class HistoryServerArchiveFetcherTest {
assertThat(overview.getJobs().iterator().next().getJobId()).isEqualTo(job2);
}
+ @TestTemplate
+ void testUpdateJobOverviewSkipsMalformedEntryInsteadOfFailingEverything()
throws Exception {
+ JobID goodJob = JobID.generate();
+ createJobArchive(remoteArchiveRootPath, goodJob, true);
+
+ HistoryServerArchiveFetcher<?> fetcher =
+ createArchiveFetcher(remoteArchiveRootPath, true,
archiveStorage);
+ fetcher.fetchArchives(EAGER);
+
+ // inject a malformed per-job overview entry alongside the good one
+ archiveStorage.putArchiveContent("overviews/malformed-job.json", "{not
valid json");
+
+ fetcher.updateJobOverview();
+
+ Object overviewObject = archiveStorage.getEntry("jobs/overview.json");
+ String overviewContent =
archiveStorage.readArchiveContent(overviewObject);
+ MultipleJobsDetails overview =
+ OBJECT_MAPPER.readValue(overviewContent,
MultipleJobsDetails.class);
+
+ assertThat(overview.getJobs()).hasSize(1);
+
assertThat(overview.getJobs().iterator().next().getJobId()).isEqualTo(goodJob);
+ }
+
+ @TestTemplate
+ void
testUpdateJobOverviewDoesNotWipeGoodOverviewWhenAllEntriesAreMalformed() throws
Exception {
+ JobID job = JobID.generate();
+ createJobArchive(remoteArchiveRootPath, job, true);
+
+ HistoryServerArchiveFetcher<?> fetcher =
+ createArchiveFetcher(remoteArchiveRootPath, true,
archiveStorage);
+ fetcher.fetchArchives(EAGER);
+
+ Object overviewObjectBefore =
archiveStorage.getEntry("jobs/overview.json");
+ String overviewContentBefore =
archiveStorage.readArchiveContent(overviewObjectBefore);
+
+ // corrupt the only per-job overview entry, simulating e.g. an
incompatible archive
+ // written by a different Flink version
+ archiveStorage.putArchiveContent("overviews/" + job + ".json", "{not
valid json anymore");
+
+ fetcher.updateJobOverview();
+
+ // the previously written combined overview must be preserved, not
replaced by an empty
+ // one
+ Object overviewObjectAfter =
archiveStorage.getEntry("jobs/overview.json");
+ String overviewContentAfter =
archiveStorage.readArchiveContent(overviewObjectAfter);
+ assertThat(overviewContentAfter).isEqualTo(overviewContentBefore);
+ }
+
@TestTemplate
void testLegacyJobOverviewMigration() throws Exception {
JobID jobId = createLegacyArchive(remoteArchiveRootPath.toPath(),
false);
