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 632551c8eb32d6..2c17b25cbeb8da 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 @@ void updateJobOverview() { try { Collection allJobs = new ArrayList<>(); List 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 612d156f244206..1cd209e96cc038 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 @@ -341,6 +341,54 @@ void testUpdateJobOverview() throws Exception { 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);