From 67ac6ed11ba1ec83abd444c9d2aff8322c13b2ce Mon Sep 17 00:00:00 2001 From: Archit Goyal Date: Sun, 23 Aug 2026 23:36:42 -0700 Subject: [PATCH] [FLINK-XXXXX][historyserver] Decouple remote archive retention from local processing limit Adds historyserver.archive.retain-remote-beyond-local-limit (default false, backward compatible). When enabled, job archives beyond historyserver.archive.retained-jobs are no longer polled/processed locally, but are kept in the remote archive directory instead of being deleted. Such archives remain reachable on demand via the existing lazyFetchArchiveProactively on-demand fetch path when historyserver.archive.load.mode is set to LAZY. This closes a gap left after FLINK-39911/FLINK-40097 introduced the pluggable ArchiveStorage backend and on-demand lazy archive loading: retainedStrategy.shouldRetain() still gated both local processing and remote deletion together, so operators could not keep an unbounded remote archive history while only actively polling/caching a small recent window locally. - HistoryServerOptions: new HISTORY_SERVER_RETAIN_REMOTE_BEYOND_LOCAL_LIMIT option. - HistoryServerArchiveFetcher: new constructor overload taking the flag; scanArchives() now routes archives beyond the retained limit to a new cleanupLocalArchivesBeyondRetainedLimit() (local-only cleanup) instead of cleanupArchivesBeyondRetainedLimit() (local+remote) when enabled. - HistoryServer: reads and wires the new option into the job archive fetcher. - Regenerated docs/layouts/shortcodes/generated/history_server_configuration.html. - Added HistoryServerArchiveFetcherTest coverage for both the default (remote-deleted) and opted-in (remote-retained, still fetchable on-demand) behavior. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../history_server_configuration.html | 6 ++ .../configuration/HistoryServerOptions.java | 30 +++++++ .../webmonitor/history/HistoryServer.java | 5 +- .../history/HistoryServerArchiveFetcher.java | 65 +++++++++++++- .../HistoryServerArchiveFetcherTest.java | 90 +++++++++++++++++++ 5 files changed, 193 insertions(+), 3 deletions(-) diff --git a/docs/layouts/shortcodes/generated/history_server_configuration.html b/docs/layouts/shortcodes/generated/history_server_configuration.html index ef521d5f999b58..415a106f9ab533 100644 --- a/docs/layouts/shortcodes/generated/history_server_configuration.html +++ b/docs/layouts/shortcodes/generated/history_server_configuration.html @@ -38,6 +38,12 @@

Enum

The mode that HistoryServer loads archives.

Possible values: + +
historyserver.archive.retain-remote-beyond-local-limit
+ false + Boolean + Whether job archives beyond the limit configured by historyserver.archive.retained-jobs should still be retained in the remote archive directory defined by historyserver.archive.fs.dir, instead of being deleted. When enabled, such archives are no longer polled/processed locally, but remain fetchable on demand when historyserver.archive.load.mode is set to LAZY. This option has no effect unless historyserver.archive.retained-jobs is set to a value other than -1. +
historyserver.archive.retained-applications
-1 diff --git a/flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java index 05de6dee6ba37a..3edf2b0c005819 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java @@ -178,6 +178,36 @@ public class HistoryServerOptions { .text(LEGACY_NOTE_MESSAGE) .build()); + /** + * If this option is enabled, job archives that fall outside {@link + * #HISTORY_SERVER_RETAINED_JOBS} are no longer processed/refreshed locally, but are kept in + * the remote archive directory instead of being deleted. They remain reachable on demand + * (e.g. by directly requesting {@code /jobs/<jobId>} in {@link + * HistoryServerArchiveLoadMode#LAZY} mode). + */ + public static final ConfigOption HISTORY_SERVER_RETAIN_REMOTE_BEYOND_LOCAL_LIMIT = + key("historyserver.archive.retain-remote-beyond-local-limit") + .booleanType() + .defaultValue(false) + .withDescription( + Description.builder() + .text( + "Whether job archives beyond the limit configured by %s should still be " + + "retained in the remote archive directory defined by %s, instead of being " + + "deleted. ", + code(HISTORY_SERVER_RETAINED_JOBS_KEY), + code(HISTORY_SERVER_ARCHIVE_DIRS.key())) + .text( + "When enabled, such archives are no longer polled/processed locally, but remain " + + "fetchable on demand when %s is set to %s. ", + code("historyserver.archive.load.mode"), + code("LAZY")) + .text( + "This option has no effect unless %s is set to a value other than %s. ", + code(HISTORY_SERVER_RETAINED_JOBS_KEY), + code("-1")) + .build()); + /** * If this option is enabled then deleted application archives are also deleted from * HistoryServer. diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java index 151fd43b8233f3..84ed9554499a17 100644 --- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java +++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java @@ -291,6 +291,8 @@ public HistoryServer( config.get(HISTORY_SERVER_LAZY_FETCH_EXECUTOR_COMMON_POOL_SIZE); int lazyFetchExecutorIndividualPoolSize = config.get(HISTORY_SERVER_LAZY_FETCH_EXECUTOR_INDIVIDUAL_POOL_SIZE); + boolean retainRemoteBeyondLocalLimit = + config.get(HistoryServerOptions.HISTORY_SERVER_RETAIN_REMOTE_BEYOND_LOCAL_LIMIT); archiveFetcher = new HistoryServerArchiveFetcher<>( refreshDirs, @@ -301,7 +303,8 @@ public HistoryServer( archiveStorage, archiveMetaInfoCache, lazyFetchExecutorCommonPoolSize, - lazyFetchExecutorIndividualPoolSize); + lazyFetchExecutorIndividualPoolSize, + retainRemoteBeyondLocalLimit); applicationArchiveFetcher = new HistoryServerApplicationArchiveFetcher<>( refreshDirs, 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..7a11129a40d9e0 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 @@ -135,6 +135,13 @@ public ArchiveEventType getType() { protected final ArchiveStorage archiveStorage; + /** + * Whether archives beyond {@link HistoryServerOptions#HISTORY_SERVER_RETAINED_JOBS} should be + * retained in the remote archive directory instead of being deleted. When {@code true}, such + * archives are only skipped for local processing, not deleted remotely. + */ + private final boolean retainRemoteBeyondLocalLimit; + /** Executor for loading archives. */ private final ExecutorService commonFetchExecutor; @@ -154,10 +161,36 @@ public ArchiveEventType getType() { int lazyFetchExecutorCommonPoolSize, int lazyFetchExecutorIndividualPoolSize) throws IOException { + this( + refreshDirs, + webDir, + archiveEventListener, + cleanupExpiredArchives, + retainedStrategy, + archiveStorage, + archiveMetaInfoCache, + lazyFetchExecutorCommonPoolSize, + lazyFetchExecutorIndividualPoolSize, + false); + } + + HistoryServerArchiveFetcher( + List refreshDirs, + File webDir, + Consumer archiveEventListener, + boolean cleanupExpiredArchives, + ArchiveRetainedStrategy retainedStrategy, + ArchiveStorage archiveStorage, + ConcurrentHashMap archiveMetaInfoCache, + int lazyFetchExecutorCommonPoolSize, + int lazyFetchExecutorIndividualPoolSize, + boolean retainRemoteBeyondLocalLimit) + throws IOException { this.refreshDirs = checkNotNull(refreshDirs); this.archiveEventListener = archiveEventListener; this.processExpiredArchiveDeletion = cleanupExpiredArchives; this.retainedStrategy = checkNotNull(retainedStrategy); + this.retainRemoteBeyondLocalLimit = retainRemoteBeyondLocalLimit; this.cachedArchivesPerRefreshDirectory = new HashMap<>(); for (HistoryServer.RefreshLocation refreshDir : refreshDirs) { cachedArchivesPerRefreshDirectory.put(refreshDir.getPath(), new HashSet<>()); @@ -240,9 +273,16 @@ void scanArchives( && processExpiredArchiveDeletion) { events.addAll(cleanupExpiredArchives(archivesToRemove)); } - // clean remote and local if (!archivesBeyondRetainedLimit.isEmpty()) { - events.addAll(cleanupArchivesBeyondRetainedLimit(archivesBeyondRetainedLimit)); + if (retainRemoteBeyondLocalLimit) { + // clean local only; the remote archive is left in place and remains + // fetchable on demand (e.g. via LAZY archive load mode). + events.addAll( + cleanupLocalArchivesBeyondRetainedLimit(archivesBeyondRetainedLimit)); + } else { + // clean remote and local + events.addAll(cleanupArchivesBeyondRetainedLimit(archivesBeyondRetainedLimit)); + } } if (!events.isEmpty()) { updateOverview(); @@ -368,6 +408,27 @@ List cleanupArchivesBeyondRetainedLimit(Map> archi return cleanupExpiredArchives(allArchiveIdsToRemove); } + /** + * Cleans up archives beyond {@link HistoryServerOptions#HISTORY_SERVER_RETAINED_JOBS} from + * the local cache only. Unlike {@link #cleanupArchivesBeyondRetainedLimit}, the remote archive + * is left untouched so that it remains fetchable on demand, e.g. via {@link + * HistoryServerOptions.HistoryServerArchiveLoadMode#LAZY} mode. + */ + List cleanupLocalArchivesBeyondRetainedLimit( + Map> archivesToRemove) { + Map> allArchiveIdsToRemove = new HashMap<>(); + + for (Map.Entry> pathSetEntry : archivesToRemove.entrySet()) { + HashSet archiveIdsToRemove = new HashSet<>(); + for (Path archive : pathSetEntry.getValue()) { + archiveIdsToRemove.add(archive.getName()); + } + allArchiveIdsToRemove.put(pathSetEntry.getKey(), archiveIdsToRemove); + } + + return cleanupExpiredArchives(allArchiveIdsToRemove); + } + void deleteFromRemote(Path archive) throws IOException { archive.getFileSystem().delete(archive, false); } 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..de153c631ce504 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 @@ -125,6 +125,34 @@ private HistoryServerArchiveFetcher createArchiveFetcher( 4); } + /** + * Create {@link HistoryServerArchiveFetcher} instance with a custom retention strategy and the + * {@code retainRemoteBeyondLocalLimit} flag, used to test the decoupling of local processing + * from remote archive retention. + */ + private HistoryServerArchiveFetcher createArchiveFetcher( + File refreshDir, + boolean cleanupExpiredJobs, + ArchiveStorage storage, + org.apache.flink.runtime.webmonitor.history.retaining.ArchiveRetainedStrategy + retainedStrategy, + boolean retainRemoteBeyondLocalLimit) + throws Exception { + List refreshDirs = + Collections.singletonList(createRefreshLocation(refreshDir)); + return new HistoryServerArchiveFetcher<>( + refreshDirs, + localArchiveRootPath, + event -> archiveEvents.add(event), + cleanupExpiredJobs, + retainedStrategy, + storage, + archiveMetaInfoCache, + 4, + 4, + retainRemoteBeyondLocalLimit); + } + // ========================================================================= // EAGER MODE TESTS // ========================================================================= @@ -369,6 +397,68 @@ void testScanArchivesWithoutFetch() throws Exception { assertThat(archiveStorage.exists("overviews/" + jobId + ".json")).isFalse(); } + @TestTemplate + void testArchivesBeyondRetainedLimitAreDeletedFromRemoteByDefault() throws Exception { + JobID retainedJobId = JobID.generate(); + JobID beyondLimitJobId = JobID.generate(); + Path beyondLimitArchivePath = + createJobArchive(remoteArchiveRootPath, beyondLimitJobId, true); + createJobArchive(remoteArchiveRootPath, retainedJobId, true); + + // retain only the archive belonging to retainedJobId, regardless of file ordering + org.apache.flink.runtime.webmonitor.history.retaining.ArchiveRetainedStrategy + retainOnlyRetainedJob = + (file, index) -> file.getPath().getName().equals(retainedJobId.toString()); + + HistoryServerArchiveFetcher fetcher = + createArchiveFetcher( + remoteArchiveRootPath, + true, + archiveStorage, + retainOnlyRetainedJob, + false); + + fetcher.fetchArchives(EAGER); + + assertThat(beyondLimitArchivePath.getFileSystem().exists(beyondLimitArchivePath)) + .as("archive beyond the retained limit should be deleted from remote by default") + .isFalse(); + } + + @TestTemplate + void testArchivesBeyondRetainedLimitAreKeptRemotelyWhenConfigured() throws Exception { + JobID retainedJobId = JobID.generate(); + JobID beyondLimitJobId = JobID.generate(); + Path beyondLimitArchivePath = + createJobArchive(remoteArchiveRootPath, beyondLimitJobId, true); + createJobArchive(remoteArchiveRootPath, retainedJobId, true); + + // retain only the archive belonging to retainedJobId, regardless of file ordering + org.apache.flink.runtime.webmonitor.history.retaining.ArchiveRetainedStrategy + retainOnlyRetainedJob = + (file, index) -> file.getPath().getName().equals(retainedJobId.toString()); + + HistoryServerArchiveFetcher fetcher = + createArchiveFetcher( + remoteArchiveRootPath, true, archiveStorage, retainOnlyRetainedJob, true); + + fetcher.fetchArchives(EAGER); + + // remote archive beyond the limit must still exist ... + assertThat(beyondLimitArchivePath.getFileSystem().exists(beyondLimitArchivePath)) + .as( + "archive beyond the retained limit must not be deleted from remote when " + + "retainRemoteBeyondLocalLimit is enabled") + .isTrue(); + // ... but must not have been processed/cached locally + assertThat(archiveStorage.exists("overviews/" + beyondLimitJobId + ".json")).isFalse(); + + // and it must still be fetchable on demand + fetcher.lazyFetchArchiveProactively(beyondLimitJobId.toString(), beyondLimitArchivePath); + waitForArchiveLoaded(archiveMetaInfoCache, beyondLimitJobId.toString()); + assertThat(archiveStorage.exists("overviews/" + beyondLimitJobId + ".json")).isTrue(); + } + @TestTemplate void testLazyFetchArchiveProactively() throws Exception { // with explicit path