[FLINK-40530][runtime-web] Isolate per-entry failures in job overview aggregation

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java

@@ -505,19 +505,42 @@ void updateJobOverview() {
         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(

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);