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/main/java/org/apache/flink/runtime/webmonitor/history/retaining/ArchiveRetainedStrategy.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/retaining/ArchiveRetainedStrategy.java
index f2e50dd73c551e..12942d4908b4bd 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/retaining/ArchiveRetainedStrategy.java
+++ b/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.
+ *
+ * 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;
+ }
}
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/retaining/CompositeArchiveRetainedStrategy.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/retaining/CompositeArchiveRetainedStrategy.java
index 2a38dfaeff773e..c66905d4433885 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/retaining/CompositeArchiveRetainedStrategy.java
+++ b/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();
}
}
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..9278f01ad5a70f 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,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