argoyal2212 commented on code in PR #29008:
URL: https://github.com/apache/flink/pull/29008#discussion_r4471562759
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -669,6 +731,42 @@ 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) {
+ if (!isOwnedByRefreshDirectory(jobId)) {
+ return;
+ }
+ 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));
+ }
+ }
+ }
+ if (!events.isEmpty()) {
+ updateOverview();
+ events.forEach(archiveEventListener);
+ }
+ }
+
+ /** Application-owned jobs are nested deeper and managed by the
application fetcher. */
+ private boolean isOwnedByRefreshDirectory(String jobId) {
+ ArchiveMetaInfo metaInfo = archiveMetaInfoCache.get(jobId);
+ if (metaInfo == null || metaInfo.getArchivePath() == null) {
+ return false;
+ }
+ Path parent = metaInfo.getArchivePath().getParent();
+ return refreshDirs.stream().anyMatch(dir ->
dir.getPath().equals(parent));
Review Comment:
Good find, both paths are now qualified with the refresh dir's filesystem
before comparing, and there is a test with an unqualified dir. Fixed in c450802.
##########
flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java:
##########
@@ -420,6 +456,134 @@ void testScanArchivesWithoutFetch() throws Exception {
assertThat(archiveStorage.exists("overviews/" + jobId +
".json")).isFalse();
}
+ @TestTemplate
+ void testArchivesBeyondCachedLimitAreOnlyCachedLocallyWithinLimit() throws
Exception {
Review Comment:
Added testLazyModeViewingAnotherJobDoesNotCancelInFlightFetch in
AbstractHistoryServerHandlerTest, it goes through the real LAZY handler and
failed with a 500 before the fix.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
