spuru9 commented on code in PR #29008:
URL: https://github.com/apache/flink/pull/29008#discussion_r4496592558
##########
flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java:
##########
@@ -420,6 +456,220 @@ 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);
+ assertThat(isCachedLocally(older)).isFalse();
+ view(fetcher, older, olderArchive);
+ assertThat(isCachedLocally(older)).isTrue();
+
+ archiveEvents.clear();
+ fetcher.fetchArchives(EAGER);
+
+ assertThat(isCachedLocally(newest)).isTrue();
+ assertThat(isCachedLocally(older)).isTrue();
+ assertThat(archiveEvents)
+ .noneMatch(
+ e ->
+ e.getId().equals(older.toString())
+ && e.getType()
+ ==
HistoryServerArchiveFetcher.ArchiveEventType
+ .DELETED);
+ }
+
+ @TestTemplate
+ void testViewingApplicationJobsDoesNotEvictThem() throws Exception {
+ JobID appJobA = JobID.generate();
+ JobID appJobB = JobID.generate();
+ fetcher = createArchiveFetcher(RETAIN_ALL, 1, 1);
+ for (JobID id : Arrays.asList(appJobA, appJobB)) {
+ archiveStorage.putArchiveContent("overviews/" + id + ".json",
"{}");
+ archiveMetaInfoCache.put(
+ id.toString(),
+ new ArchiveMetaInfo(
+ id.toString(),
+ CREATED,
+ new Path(
+ remoteArchiveRootPath.toURI().toString(),
+ "cluster/applications/app/jobs/" + id)));
+ }
+
+ fetcher.markRecentlyViewed(appJobA.toString());
+ fetcher.markRecentlyViewed(appJobB.toString());
+
+ assertThat(isCachedLocally(appJobA)).isTrue();
+ assertThat(isCachedLocally(appJobB)).isTrue();
+ assertThat(archiveMetaInfoCache).containsKeys(appJobA.toString(),
appJobB.toString());
+ }
+
+ @TestTemplate
+ void testLeastRecentlyViewedArchiveIsEvictedLocally() throws Exception {
+ JobID newest = JobID.generate();
+ JobID old1 = JobID.generate();
+ JobID old2 = JobID.generate();
+ Path old1Archive = createArchive(old1, 1000);
+ Path old2Archive = createArchive(old2, 1500);
+ createArchive(newest, 2000);
+
+ fetcher = createArchiveFetcher(RETAIN_ALL, 1, 1);
+ fetcher.fetchArchives(EAGER);
+ view(fetcher, old1, old1Archive);
+ view(fetcher, old2, old2Archive);
+
+ assertThat(isCachedLocally(old1)).isFalse();
+ assertThat(isCachedLocally(old2)).isTrue();
+ assertThat(old1Archive.getFileSystem().exists(old1Archive)).isTrue();
+ }
+
+ @TestTemplate
+ void testViewingJobInsideCachedWindowDoesNotTakeViewedSlot() throws
Exception {
+ JobID newest = JobID.generate();
+ JobID old = JobID.generate();
+ Path oldArchive = createArchive(old, 1000);
+ Path newestArchive = createArchive(newest, 2000);
+
+ fetcher = createArchiveFetcher(RETAIN_ALL, 1, 1);
+ fetcher.fetchArchives(EAGER);
+ view(fetcher, old, oldArchive);
+ view(fetcher, newest, newestArchive);
+
+ assertThat(isCachedLocally(old)).isTrue();
+ assertThat(isCachedLocally(newest)).isTrue();
+ }
+
+ @TestTemplate
+ void testViewedArchivesAreNotEvictedWhenCachedLimitIsNotSet() throws
Exception {
+ JobID first = JobID.generate();
+ JobID second = JobID.generate();
+ Path firstArchive = createArchive(first, 1000);
+ Path secondArchive = createArchive(second, 2000);
+
+ fetcher = createArchiveFetcher(RETAIN_ALL, -1, 1);
+ view(fetcher, first, firstArchive);
+ view(fetcher, second, secondArchive);
+
+ assertThat(isCachedLocally(first)).isTrue();
+ assertThat(isCachedLocally(second)).isTrue();
+ }
+
+ @TestTemplate
+ void testArchiveWithFetchInFlightIsNotEvictedByViewingAnotherJob() throws
Exception {
+ JobID newest = JobID.generate();
+ JobID inFlight = JobID.generate();
+ JobID other = JobID.generate();
+ Path inFlightArchive = createArchive(inFlight, 1000);
+ Path otherArchive = createArchive(other, 1500);
+ createArchive(newest, 2000);
+
+ fetcher = createArchiveFetcher(RETAIN_ALL, 1, 1);
+ fetcher.fetchArchives(EAGER);
+ view(fetcher, inFlight, inFlightArchive);
+
archiveMetaInfoCache.get(inFlight.toString()).setEventType(DETAIL_PARSING);
+
+ view(fetcher, other, otherArchive);
+ assertThat(isCachedLocally(inFlight)).isTrue();
+
+ // once the fetch is done, the next view evicts it
+ archiveMetaInfoCache.get(inFlight.toString()).setEventType(CREATED);
+ fetcher.markRecentlyViewed(other.toString());
+ assertThat(isCachedLocally(inFlight)).isFalse();
+ assertThat(isCachedLocally(other)).isTrue();
+ }
+
+ @TestTemplate
+ void testViewedArchiveIsTrackedWhenRefreshDirIsNotQualified() throws
Exception {
+ JobID newest = JobID.generate();
+ JobID older = JobID.generate();
+ Path olderArchive = createArchive(older, 1000);
+ createArchive(newest, 2000);
+
+ HistoryServer.RefreshLocation location =
createRefreshLocation(remoteArchiveRootPath);
+ Path unqualified = new Path(remoteArchiveRootPath.getAbsolutePath());
Review Comment:
minor: This test also passes on the code before the fix. `older` is already
outside the cached window at the first scan, so the scan never caches or evicts
it whether it's tracked or not. Could it view two old jobs instead and assert
the first one gets evicted? That fails without `makeQualified`.
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -669,6 +737,75 @@ 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 (cachedRetainedJobs <= 0 || !isOwnedByRefreshDirectory(jobId) ||
isCachedByScan(jobId)) {
+ return;
+ }
+ List<ArchiveEvent> events = new ArrayList<>();
+ synchronized (mostRecentlyViewedJobs) {
+ mostRecentlyViewedJobs.put(jobId, Boolean.TRUE);
+ Iterator<String> eldest =
mostRecentlyViewedJobs.keySet().iterator();
+ while (mostRecentlyViewedJobs.size() >
numCachedMostRecentlyViewedJobs
+ && eldest.hasNext()) {
+ String evictedId = eldest.next();
+ // an in-flight fetch must finish first, a later call evicts it
+ if (evictedId.equals(jobId) || isFetchInFlight(evictedId)) {
+ continue;
+ }
+ eldest.remove();
+ // archives that entered the cached window are managed by the
scan
+ if (!isCachedByScan(evictedId)) {
+ events.add(deleteJobFiles(evictedId));
+ }
+ }
+ }
+ if (!events.isEmpty()) {
+ updateOverview();
+ events.forEach(archiveEventListener);
+ }
+ }
+
+ private boolean isCachedByScan(String jobId) {
+ return cachedArchivesPerRefreshDirectory.values().stream()
+ .anyMatch(ids -> ids.contains(jobId));
+ }
+
+ private boolean isFetchInFlight(String jobId) {
+ if (commonFetchTasks.containsKey(jobId) ||
individualFetchTasks.containsKey(jobId)) {
Review Comment:
minor: The fetch task sets CREATED before the finally removes it from
fetchTasks, and individual fetches run updateOverview() in between, so a job
that just finished still counts as in flight here.
testLeastRecentlyViewedArchiveIsEvictedLocally fails intermittently for me
because of it when run on its own. Could the task drop out of the map before
setting CREATED, or could this just check the event type?
--
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]
