[FLINK-37155][historyserver] Decouple remote archive retention from local processing limit

docs/layouts/shortcodes/generated/history_server_configuration.html

@@ -38,6 +38,12 @@
             <td><p>Enum</p></td>
             <td>The mode that HistoryServer loads archives.<br /><br />Possible values:<ul><li>"EAGER"</li><li>"LAZY"</li></ul></td>
         </tr>
+        <tr>
+            <td><h5>historyserver.archive.retain-remote-beyond-local-limit</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether job archives beyond the limit configured by <code class="highlighter-rouge">historyserver.archive.retained-jobs</code> should still be retained in the remote archive directory defined by <code class="highlighter-rouge">historyserver.archive.fs.dir</code>, instead of being deleted. When enabled, such archives are no longer polled/processed locally, but remain fetchable on demand when <code class="highlighter-rouge">historyserver.archive.load.mode</code> is set to <code class="highlighter-rouge">LAZY</code>. This option has no effect unless <code class="highlighter-rouge">historyserver.archive.retained-jobs</code> is set to a value other than <code class="highlighter-rouge">-1</code>. </td>
+        </tr>
         <tr>
             <td><h5>historyserver.archive.retained-applications</h5></td>
             <td style="word-wrap: break-word;">-1</td>

flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java

@@ -178,6 +178,34 @@ 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<Boolean> 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.

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

@@ -291,6 +291,11 @@ 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);
+        LOG.info(
+                "Archives beyond the local retention limit will {} in the remote archive directory.",
+                retainRemoteBeyondLocalLimit ? "be retained" : "be deleted");
         archiveFetcher =
                 new HistoryServerArchiveFetcher<>(
                         refreshDirs,
@@ -301,7 +306,8 @@ public HistoryServer(
                         archiveStorage,
                         archiveMetaInfoCache,
                         lazyFetchExecutorCommonPoolSize,
-                        lazyFetchExecutorIndividualPoolSize);
+                        lazyFetchExecutorIndividualPoolSize,
+                        retainRemoteBeyondLocalLimit);
         applicationArchiveFetcher =
                 new HistoryServerApplicationArchiveFetcher<>(
                         refreshDirs,

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

@@ -135,6 +135,13 @@ public ArchiveEventType getType() {
 
     protected final ArchiveStorage<Entry> 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<HistoryServer.RefreshLocation> refreshDirs,
+            File webDir,
+            Consumer<ArchiveEvent> archiveEventListener,
+            boolean cleanupExpiredArchives,
+            ArchiveRetainedStrategy retainedStrategy,
+            ArchiveStorage<Entry> archiveStorage,
+            ConcurrentHashMap<String, ArchiveMetaInfo> 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<>());
@@ -198,6 +231,7 @@ void scanArchives(
             cachedArchivesPerRefreshDirectory.forEach(
                     (path, archives) -> archivesToRemove.put(path, new HashSet<>(archives)));
             Map<Path, Set<Path>> archivesBeyondRetainedLimit = new HashMap<>();
+            Map<Path, Set<Path>> archivesExpiredByTtl = new HashMap<>();
             for (HistoryServer.RefreshLocation refreshLocation : refreshDirs) {
                 Path refreshDir = refreshLocation.getPath();
                 LOG.debug("Checking archive directory {}.", refreshDir);
@@ -223,9 +257,17 @@ void scanArchives(
 
                     fileOrderedIndexOnModifiedTime++;
                     if (!retainedStrategy.shouldRetain(archive, fileOrderedIndexOnModifiedTime)) {
-                        archivesBeyondRetainedLimit
-                                .computeIfAbsent(refreshDir, ignored -> new HashSet<>())
-                                .add(archivePath);
+                        if (retainedStrategy.isExpiredByTtl(archive)) {
+                            // TTL expiry always applies remote deletion, regardless of
+                            // retainRemoteBeyondLocalLimit, which only concerns the count limit.
+                            archivesExpiredByTtl
+                                    .computeIfAbsent(refreshDir, ignored -> new HashSet<>())
+                                    .add(archivePath);
+                        } else {
+                            archivesBeyondRetainedLimit
+                                    .computeIfAbsent(refreshDir, ignored -> new HashSet<>())
+                                    .add(archivePath);
+                        }
                         continue;
                     }
 
@@ -240,9 +282,20 @@ void scanArchives(
                     && processExpiredArchiveDeletion) {
                 events.addAll(cleanupExpiredArchives(archivesToRemove));
             }
-            // clean remote and local
+            if (!archivesExpiredByTtl.isEmpty()) {
+                // clean remote and local; TTL expiry is unaffected by retainRemoteBeyondLocalLimit
+                events.addAll(cleanupArchivesBeyondRetainedLimit(archivesExpiredByTtl));
+            }
             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 +421,27 @@ List<ArchiveEvent> cleanupArchivesBeyondRetainedLimit(Map<Path, Set<Path>> 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<ArchiveEvent> cleanupLocalArchivesBeyondRetainedLimit(
+            Map<Path, Set<Path>> archivesToRemove) {
+        Map<Path, Set<String>> allArchiveIdsToRemove = new HashMap<>();
+
+        for (Map.Entry<Path, Set<Path>> pathSetEntry : archivesToRemove.entrySet()) {
+            HashSet<String> 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);
     }

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

@@ -31,4 +31,19 @@ public interface ArchiveRetainedStrategy {
      * @return The result that indicates whether the file should be retained.
      */
     boolean shouldRetain(FileStatus file, int fileOrderedIndex);
+
+    /**
+     * Judge whether the file is rejected specifically because it has exceeded its configured
+     * time-to-live, as opposed to being rejected by a count-based retention limit.
+     *
+     * <p>This allows callers that want to treat count-limit rejections differently from TTL expiry
+     * (e.g. to only stop archiving locally without affecting TTL-based remote deletion) to
+     * distinguish the two cases.
+     *
+     * @param file the target file to judge.
+     * @return {@code true} if the file is rejected due to TTL expiry.
+     */
+    default boolean isExpiredByTtl(FileStatus file) {
+        return false;
+    }
 }

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

@@ -80,6 +80,11 @@ public boolean shouldRetain(FileStatus file, int fileOrderedIndex) {
         }
         return strategies.stream().allMatch(s -> s.shouldRetain(file, fileOrderedIndex));
     }
+
+    @Override
+    public boolean isExpiredByTtl(FileStatus file) {
+        return strategies.stream().anyMatch(s -> s.isExpiredByTtl(file));
+    }
 }
 
 /** The time to live based retained strategy. */
@@ -98,10 +103,15 @@ class TimeToLiveArchiveRetainedStrategy implements ArchiveRetainedStrategy {
 
     @Override
     public boolean shouldRetain(FileStatus file, int fileOrderedIndex) {
+        return !isExpiredByTtl(file);
+    }
+
+    @Override
+    public boolean isExpiredByTtl(FileStatus file) {
         if (ttlThreshold == null) {
-            return true;
+            return false;
         }
-        return Instant.now().toEpochMilli() - file.getModificationTime() < ttlThreshold.toMillis();
+        return Instant.now().toEpochMilli() - file.getModificationTime() >= ttlThreshold.toMillis();
     }
 }
 

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<HistoryServer.RefreshLocation> 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,104 @@ 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 testTtlExpiredArchivesAreDeletedFromRemoteEvenWhenBeyondLimitIsKept() throws Exception {
+        JobID retainedJobId = JobID.generate();
+        JobID ttlExpiredJobId = JobID.generate();
+        Path ttlExpiredArchivePath = createJobArchive(remoteArchiveRootPath, ttlExpiredJobId, true);
+        createJobArchive(remoteArchiveRootPath, retainedJobId, true);
+
+        // reject the ttlExpiredJobId archive, but mark it as TTL-expired rather than beyond
+        // the count limit
+        org.apache.flink.runtime.webmonitor.history.retaining.ArchiveRetainedStrategy
+                ttlExpiringStrategy =
+                        new org.apache.flink.runtime.webmonitor.history.retaining
+                                .ArchiveRetainedStrategy() {
+                            @Override
+                            public boolean shouldRetain(
+                                    org.apache.flink.core.fs.FileStatus file, int index) {
+                                return file.getPath().getName().equals(retainedJobId.toString());
+                            }
+
+                            @Override
+                            public boolean isExpiredByTtl(
+                                    org.apache.flink.core.fs.FileStatus file) {
+                                return file.getPath().getName().equals(ttlExpiredJobId.toString());
+                            }
+                        };
+
+        HistoryServerArchiveFetcher<?> fetcher =
+                createArchiveFetcher(
+                        remoteArchiveRootPath, true, archiveStorage, ttlExpiringStrategy, true);
+
+        fetcher.fetchArchives(EAGER);
+
+        // TTL-expired archives must be deleted remotely regardless of retainRemoteBeyondLocalLimit
+        assertThat(ttlExpiredArchivePath.getFileSystem().exists(ttlExpiredArchivePath))
+                .as(
+                        "TTL-expired archive must be deleted from remote even when "
+                                + "retainRemoteBeyondLocalLimit is enabled")
+                .isFalse();
+    }
+
     @TestTemplate
     void testLazyFetchArchiveProactively() throws Exception {
         // with explicit path