You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "argoyal2212 (via GitHub)" <gi...@apache.org> on 2026/08/24 06:39:51 UTC
[PR] [FLINK-XXXXX][historyserver] Decouple remote archive retention from local processing limit [flink]
argoyal2212 opened a new pull request, #29008:
URL: https://github.com/apache/flink/pull/29008
## What is the purpose of the change
`HistoryServerOptions.HISTORY_SERVER_RETAINED_JOBS` currently gates both (a) which archives the `HistoryServer` polls/processes locally, and (b) which archives are deleted from the remote archive directory. This means operators cannot keep a large/unbounded remote job-archive history while only actively refreshing a small recent window locally — trimming the local processing window also deletes the "older" jobs remotely.
This closes that gap, which is what FLIP-505 originally set out to solve, before FLIP-584/585 (FLINK-39911, FLINK-40097) landed a more general pluggable `ArchiveStorage` backend + on-demand lazy-fetch mechanism that already covers FLIP-505's other goals (per-job on-demand fetch, avoiding local disk/inode exhaustion via a pluggable storage backend). This PR adds the one remaining piece: decoupling remote retention from the local processing limit.
## Brief change log
- Added `historyserver.archive.retain-remote-beyond-local-limit` (boolean, default `false`, fully backward compatible).
- `HistoryServerArchiveFetcher.scanArchives()`: when enabled, archives beyond `historyserver.archive.retained-jobs` are cleaned up locally only (new `cleanupLocalArchivesBeyondRetainedLimit`), leaving the remote archive intact, instead of calling `cleanupArchivesBeyondRetainedLimit` (which deletes both local and remote).
- Such archives remain reachable on demand via the existing `lazyFetchArchiveProactively` path when `historyserver.archive.load.mode=LAZY`.
- `HistoryServer` reads and wires the new option into the job-archive fetcher (applications are unaffected — this only applies to job archives, matching FLIP-505's original scope).
- Regenerated `docs/layouts/shortcodes/generated/history_server_configuration.html`.
## Verifying this change
This change added tests and can be verified as follows:
- Added `HistoryServerArchiveFetcherTest#testArchivesBeyondRetainedLimitAreDeletedFromRemoteByDefault` and `#testArchivesBeyondRetainedLimitAreKeptRemotelyWhenConfigured`, covering both storage backends (File/RocksDB), asserting: default behavior is unchanged (remote archive deleted beyond limit); with the flag enabled, the remote archive persists, is not locally cached, and is still fetchable on demand.
- All 24 tests in `HistoryServerArchiveFetcherTest` pass locally.
## Does this pull request potentially affect one of the following parts?
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: yes, one new `@PublicEvolving` `ConfigOption`
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? JavaDocs + generated configuration docs
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-XXXXX][historyserver] Decouple remote archive retention from local processing limit [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29008:
URL: https://github.com/apache/flink/pull/29008#issuecomment-5391703194
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67ac6ed11ba1ec83abd444c9d2aff8322c13b2ce",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "67ac6ed11ba1ec83abd444c9d2aff8322c13b2ce",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 67ac6ed11ba1ec83abd444c9d2aff8322c13b2ce UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37155][historyserver] Decouple remote archive retention from local processing limit [flink]
Posted by "argoyal2212 (via GitHub)" <gi...@apache.org>.
argoyal2212 commented on PR #29008:
URL: https://github.com/apache/flink/pull/29008#issuecomment-5725851340
@rmetzger can you please help review this PR. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37155][historyserver] Decouple remote archive retention from local processing limit [flink]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #29008:
URL: https://github.com/apache/flink/pull/29008#discussion_r4302350218
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -669,6 +731,29 @@ void lazyFetchArchiveProactively(String jobId, @Nullable Path archivePath) throw
}
}
+ /** Marks the job as recently viewed and evicts the least recently viewed ones locally. */
+ void markRecentlyViewed(String jobId) {
+ List<ArchiveEvent> events = new ArrayList<>();
+ synchronized (mostRecentlyViewedJobs) {
+ mostRecentlyViewedJobs.put(jobId, Boolean.TRUE);
+ Iterator<String> eldest = mostRecentlyViewedJobs.keySet().iterator();
+ while (mostRecentlyViewedJobs.size() > numCachedMostRecentlyViewedJobs) {
+ String evictedId = eldest.next();
+ eldest.remove();
+ // archives still inside the cached window are managed by the scan
+ if (cachedArchivesPerRefreshDirectory.values().stream()
+ .noneMatch(ids -> ids.contains(evictedId))) {
+ cleanUpLazyFetchTask(evictedId);
+ events.add(deleteJobFiles(evictedId));
Review Comment:
In LAZY mode, this also evicts application-owned jobs, even when `cached-retained-jobs` is unset. The handler marks every requested job, but application jobs are tracked by `HistoryServerApplicationArchiveFetcher`, not this fetcher's `cachedArchivesPerRefreshDirectory`. With the default viewed-job limit of 1, viewing application jobs A then B deletes A's local files and shared metadata. Revisiting A cannot reload it: proactive fetching searches only `<refreshDir>/<jobId>`, while A is nested under an application directory, and the application fetcher skips the already-cached application. Exclude application-owned jobs from this eviction policy and add a regression test that revisits A after viewing B with default cache settings.
##########
flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java:
##########
@@ -420,6 +455,102 @@ void testScanArchivesWithoutFetch() throws Exception {
assertThat(archiveStorage.exists("overviews/" + jobId + ".json")).isFalse();
}
+ @TestTemplate
+ void testArchivesBeyondCachedLimitAreOnlyCachedLocallyWithinLimit() throws Exception {
+ JobID newest = JobID.generate();
+ JobID older = JobID.generate();
+ Path olderArchive = createArchive(older, 1000);
+ createArchive(newest, 2000);
+
+ fetcher = createArchiveFetcher(RETAIN_ALL, 1, 1);
+ fetcher.fetchArchives(EAGER);
+
+ assertThat(isCachedLocally(newest)).isTrue();
+ assertThat(isCachedLocally(older)).isFalse();
+ assertThat(olderArchive.getFileSystem().exists(olderArchive)).isTrue();
+ }
+
+ @TestTemplate
+ void testCachedArchivePushedOutOfLimitIsEvictedLocallyOnly() throws Exception {
+ JobID first = JobID.generate();
+ JobID second = JobID.generate();
+ Path firstArchive = createArchive(first, 1000);
+
+ fetcher = createArchiveFetcher(RETAIN_ALL, 1, 1);
+ fetcher.fetchArchives(EAGER);
+ assertThat(isCachedLocally(first)).isTrue();
+
+ createArchive(second, 2000);
+ archiveEvents.clear();
+ fetcher.fetchArchives(EAGER);
+
+ assertThat(isCachedLocally(first)).isFalse();
+ assertThat(isCachedLocally(second)).isTrue();
+ assertThat(firstArchive.getFileSystem().exists(firstArchive)).isTrue();
+ assertThat(archiveEvents)
+ .anyMatch(
+ e ->
+ e.getId().equals(first.toString())
+ && e.getType()
+ == HistoryServerArchiveFetcher.ArchiveEventType
+ .DELETED);
+ }
+
+ @TestTemplate
+ void testViewedArchiveBeyondLimitSurvivesNextScan() throws Exception {
+ JobID newest = JobID.generate();
+ JobID older = JobID.generate();
+ Path olderArchive = createArchive(older, 1000);
+ createArchive(newest, 2000);
+
+ fetcher = createArchiveFetcher(RETAIN_ALL, 1, 1);
+ fetcher.fetchArchives(EAGER);
+ view(fetcher, older, olderArchive);
+ assertThat(isCachedLocally(older)).isTrue();
+
+ archiveEvents.clear();
+ fetcher.fetchArchives(EAGER);
+
+ assertThat(archiveEvents).isEmpty();
+ assertThat(isCachedLocally(older)).isTrue();
Review Comment:
Both archives exist before the first scan, so `older` never enters `cachedArchivesPerRefreshDirectory`; proactive fetching does not add it either. The next scan therefore never considers it for local eviction, and this test passes even if the recently-viewed exclusion is removed. Cache and mark `older` as viewed before creating `newest`, then scan and verify that both remain cached and no deletion event is emitted for `older`. This exercises the scan's new protection.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37155][historyserver] Decouple remote archive retention from local processing limit [flink]
Posted by "argoyal2212 (via GitHub)" <gi...@apache.org>.
argoyal2212 commented on code in PR #29008:
URL: https://github.com/apache/flink/pull/29008#discussion_r4264147271
##########
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);
Review Comment:
Done. Both option values are now logged at startup in HistoryServer.
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -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));
Review Comment:
Fixed. Remote retention (count and TTL) is now checked first and deletes as before, so TTL-expired archives are never kept. Only archives that pass it can be evicted locally. The boolean flag is gone.
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java:
##########
@@ -301,7 +306,8 @@ public HistoryServer(
archiveStorage,
archiveMetaInfoCache,
lazyFetchExecutorCommonPoolSize,
- lazyFetchExecutorIndividualPoolSize);
+ lazyFetchExecutorIndividualPoolSize,
+ retainRemoteBeyondLocalLimit);
applicationArchiveFetcher =
new HistoryServerApplicationArchiveFetcher<>(
Review Comment:
Right, this only covers legacy job archives. I've said so in the option description and docs. Application support would be a separate change.
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -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;
}
Review Comment:
Fixed. Local eviction now only touches ids an earlier scan cached, and recently viewed jobs are skipped, so repeated scans do nothing.
##########
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")
Review Comment:
Switched to the voted keys, historyserver.archive.cached-retained-jobs and historyserver.archive.num-cached-most-recently-viewed-jobs, and dropped the custom one. Setting cached-retained-jobs is what decouples local from remote, as in the FLIP. I also added the most recently viewed cache here. I'll check #26878 for overlap.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37155][historyserver] Decouple remote archive retention from local processing limit [flink]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #29008:
URL: https://github.com/apache/flink/pull/29008#discussion_r4262739981
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -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);
Review Comment:
This delegates to `cleanupExpiredArchives`, which cancels both common and individual lazy-fetch futures and deletes their partially written entries. In LAZY mode the fixed-delay scan can therefore classify a remotely retained, out-of-window job while a request is fetching it on `individualFetchExecutor`; the waiting request then observes cancellation/missing content instead of the promised on-demand archive. Coordinate cleanup with active individual fetches (for example, defer cleanup for an in-flight job under synchronization and retry on a later scan) rather than cancelling the request-driven fetch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37155][historyserver] Decouple remote archive retention from local processing limit [flink]
Posted by "spuru9 (via GitHub)" <gi...@apache.org>.
spuru9 commented on code in PR #29008:
URL: https://github.com/apache/flink/pull/29008#discussion_r4262582408
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java:
##########
@@ -301,7 +306,8 @@ public HistoryServer(
archiveStorage,
archiveMetaInfoCache,
lazyFetchExecutorCommonPoolSize,
- lazyFetchExecutorIndividualPoolSize);
+ lazyFetchExecutorIndividualPoolSize,
+ retainRemoteBeyondLocalLimit);
applicationArchiveFetcher =
new HistoryServerApplicationArchiveFetcher<>(
Review Comment:
major: Since 2.3 (FLINK-38761), every job is archived under `applications/<app>/jobs/`, and the app fetcher never gets this flag. So it only covers archives from 2.2 and older JMs, and app jobs can't be fetched on demand anyway, because `lazyFetchArchiveProactively` only looks in `<refresh-dir>/<job-id>`.
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -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));
+ }
}
Review Comment:
minor: Goes with the comment on L259-272. Change L234 to `Map<Path, Set<String>> archivesToEvictLocally`; then the flag-off path matches master and `cleanupLocalArchivesBeyondRetainedLimit` can go.
```suggestion
// clean remote and local
if (!archivesBeyondRetainedLimit.isEmpty()) {
events.addAll(cleanupArchivesBeyondRetainedLimit(archivesBeyondRetainedLimit));
}
// clean local only, the remote archive is kept
if (!archivesToEvictLocally.isEmpty()) {
events.addAll(cleanupExpiredArchives(archivesToEvictLocally));
}
```
##########
flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java:
##########
@@ -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();
+ }
Review Comment:
minor: A second scan here would have caught the re-cleaning in `scanArchives`. The application layout and the real `isExpiredByTtl` aren't tested either.
```suggestion
// and it must still be fetchable on demand
fetcher.lazyFetchArchiveProactively(beyondLimitJobId.toString(), beyondLimitArchivePath);
waitForArchiveLoaded(archiveMetaInfoCache, beyondLimitJobId.toString());
assertThat(archiveStorage.exists("overviews/" + beyondLimitJobId + ".json")).isTrue();
// the next scan must neither re-emit events for it nor wipe the on-demand copy
archiveEvents.clear();
fetcher.fetchArchives(EAGER);
assertThat(archiveEvents)
.extracting(HistoryServerArchiveFetcher.ArchiveEvent::getId)
.doesNotContain(beyondLimitJobId.toString());
assertThat(archiveStorage.exists("overviews/" + beyondLimitJobId + ".json")).isTrue();
}
```
##########
flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java:
##########
@@ -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 =
Review Comment:
minor: After a rebase, these locals hide the `fetcher` field that FLINK-40567 added, so `tearDown` never closes them and the `@TempDir` cleanup race comes back. Assign the field instead (same at L413 and L484).
##########
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")
Review Comment:
major: FLIP-505 was voted with `cached-retained-jobs` and `num-cached-most-recently-viewed-jobs`, not this key. Can you get this agreed on dev@ (the "Reopening FLIP-505" thread got no replies), or switch to the voted keys? #26878 is also still open on the same JIRA.
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -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;
}
Review Comment:
major: Archives past the limit stay in the remote dir, so every scan cleans all of them again (an INFO log, 3 deletes and a `DELETED` event each). That same pass also wipes or cancels whatever was fetched on demand. Only evict what an earlier scan cached:
```suggestion
if (!retainedStrategy.shouldRetain(archive, fileOrderedIndexOnModifiedTime)) {
if (retainRemoteBeyondLocalLimit
&& !retainedStrategy.isExpiredByTtl(archive)) {
// the remote copy stays; only evict what an earlier scan cached
if (cachedArchivesPerRefreshDirectory
.get(refreshDir)
.contains(archiveId)) {
archivesToEvictLocally
.computeIfAbsent(refreshDir, ignored -> new HashSet<>())
.add(archiveId);
}
} else {
archivesBeyondRetainedLimit
.computeIfAbsent(refreshDir, ignored -> new HashSet<>())
.add(archivePath);
}
continue;
}
```
##########
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"))
Review Comment:
minor: Worth stating the cost in the description: the remote dir then grows without limit and gets listed on every refresh, so pair it with `retained-ttl`. A line in `historyserver.md` (en and zh) would help too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37155][historyserver] Decouple remote archive retention from local processing limit [flink]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #29008:
URL: https://github.com/apache/flink/pull/29008#discussion_r3850005639
##########
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);
Review Comment:
Because this boolean changes whether remote archives are deleted on refresh, its effective value should be logged when the HistoryServer initializes. Nothing currently records whether this retention behavior is enabled, leaving operators unable to confirm the effective setting from the startup logs.
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -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));
Review Comment:
This branch treats every rejection from the composite retention strategy as a count-limit overflow, but that strategy also rejects archives whose TTL has expired. With `historyserver.archive.retained-jobs=-1`, a configured `historyserver.archive.retained-ttl`, and this flag enabled, TTL-expired archives are therefore kept remotely, contradicting the option documentation that the flag has no effect when the count limit is `-1` and bypassing the existing remote TTL cleanup. Please limit the local-only path to archives rejected by the count policy, or update the documented contract and tests if retaining TTL-expired archives is intentional.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37155][historyserver] Decouple remote archive retention from local processing limit [flink]
Posted by "argoyal2212 (via GitHub)" <gi...@apache.org>.
argoyal2212 commented on code in PR #29008:
URL: https://github.com/apache/flink/pull/29008#discussion_r4264148846
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -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));
+ }
}
Review Comment:
Took this. Eviction goes through archivesToEvictLocally and the existing cleanupExpiredArchives, the flag-off path matches master, and the extra helper is removed.
##########
flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java:
##########
@@ -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();
+ }
Review Comment:
Added testViewedArchiveBeyondLimitSurvivesNextScan, which rescans and checks no events are emitted and the on-demand copy survives. I did not add a separate TTL or application layout case, since the TTL path no longer has special logic.
##########
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"))
Review Comment:
Added to the option description (unbounded remote growth, pair with retained-ttl) and added a section to historyserver.md in en and zh.
##########
flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java:
##########
@@ -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 =
Review Comment:
Fixed. The new tests assign the fetcher field, so tearDown closes them.
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -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);
Review Comment:
Scan eviction and most recently viewed updates now run under one lock, and a viewed job is never evicted by a scan. A viewed job can only be cancelled when it is the least recently viewed one beyond the configured limit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37155][historyserver] Decouple remote archive retention from local processing limit [flink]
Posted by "argoyal2212 (via GitHub)" <gi...@apache.org>.
argoyal2212 commented on PR #29008:
URL: https://github.com/apache/flink/pull/29008#issuecomment-5434630686
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37155][historyserver] Decouple remote archive retention from local processing limit [flink]
Posted by "argoyal2212 (via GitHub)" <gi...@apache.org>.
argoyal2212 commented on code in PR #29008:
URL: https://github.com/apache/flink/pull/29008#discussion_r4304060208
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -669,6 +731,29 @@ void lazyFetchArchiveProactively(String jobId, @Nullable Path archivePath) throw
}
}
+ /** Marks the job as recently viewed and evicts the least recently viewed ones locally. */
+ void markRecentlyViewed(String jobId) {
+ List<ArchiveEvent> events = new ArrayList<>();
+ synchronized (mostRecentlyViewedJobs) {
+ mostRecentlyViewedJobs.put(jobId, Boolean.TRUE);
+ Iterator<String> eldest = mostRecentlyViewedJobs.keySet().iterator();
+ while (mostRecentlyViewedJobs.size() > numCachedMostRecentlyViewedJobs) {
+ String evictedId = eldest.next();
+ eldest.remove();
+ // archives still inside the cached window are managed by the scan
+ if (cachedArchivesPerRefreshDirectory.values().stream()
+ .noneMatch(ids -> ids.contains(evictedId))) {
+ cleanUpLazyFetchTask(evictedId);
+ events.add(deleteJobFiles(evictedId));
Review Comment:
Good catch. markRecentlyViewed now only tracks jobs under this fetcher refresh dirs, so application-owned jobs are never evicted. Added a regression test. Fixed in f2e7f1c.
##########
flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java:
##########
@@ -420,6 +455,102 @@ void testScanArchivesWithoutFetch() throws Exception {
assertThat(archiveStorage.exists("overviews/" + jobId + ".json")).isFalse();
}
+ @TestTemplate
+ void testArchivesBeyondCachedLimitAreOnlyCachedLocallyWithinLimit() throws Exception {
+ JobID newest = JobID.generate();
+ JobID older = JobID.generate();
+ Path olderArchive = createArchive(older, 1000);
+ createArchive(newest, 2000);
+
+ fetcher = createArchiveFetcher(RETAIN_ALL, 1, 1);
+ fetcher.fetchArchives(EAGER);
+
+ assertThat(isCachedLocally(newest)).isTrue();
+ assertThat(isCachedLocally(older)).isFalse();
+ assertThat(olderArchive.getFileSystem().exists(olderArchive)).isTrue();
+ }
+
+ @TestTemplate
+ void testCachedArchivePushedOutOfLimitIsEvictedLocallyOnly() throws Exception {
+ JobID first = JobID.generate();
+ JobID second = JobID.generate();
+ Path firstArchive = createArchive(first, 1000);
+
+ fetcher = createArchiveFetcher(RETAIN_ALL, 1, 1);
+ fetcher.fetchArchives(EAGER);
+ assertThat(isCachedLocally(first)).isTrue();
+
+ createArchive(second, 2000);
+ archiveEvents.clear();
+ fetcher.fetchArchives(EAGER);
+
+ assertThat(isCachedLocally(first)).isFalse();
+ assertThat(isCachedLocally(second)).isTrue();
+ assertThat(firstArchive.getFileSystem().exists(firstArchive)).isTrue();
+ assertThat(archiveEvents)
+ .anyMatch(
+ e ->
+ e.getId().equals(first.toString())
+ && e.getType()
+ == HistoryServerArchiveFetcher.ArchiveEventType
+ .DELETED);
+ }
+
+ @TestTemplate
+ void testViewedArchiveBeyondLimitSurvivesNextScan() throws Exception {
+ JobID newest = JobID.generate();
+ JobID older = JobID.generate();
+ Path olderArchive = createArchive(older, 1000);
+ createArchive(newest, 2000);
+
+ fetcher = createArchiveFetcher(RETAIN_ALL, 1, 1);
+ fetcher.fetchArchives(EAGER);
+ view(fetcher, older, olderArchive);
+ assertThat(isCachedLocally(older)).isTrue();
+
+ archiveEvents.clear();
+ fetcher.fetchArchives(EAGER);
+
+ assertThat(archiveEvents).isEmpty();
+ assertThat(isCachedLocally(older)).isTrue();
Review Comment:
Agreed. The test now caches and views older before newest exists, then checks both stay cached with no DELETED event for older. It fails if the viewed protection is removed. Fixed in f2e7f1c.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37155][historyserver] Decouple remote archive retention from local processing limit [flink]
Posted by "argoyal2212 (via GitHub) via issues" <is...@flink.apache.org>.
argoyal2212 commented on PR #29008:
URL: https://github.com/apache/flink/pull/29008#issuecomment-6615881411
@RocMarshal @rmetzger can I please get a review on the PR. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org