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, so it doesn't cover
the qualification bug. On the local FS the plain path ends up equal to the
fully qualified one. Could you make the refresh dir differ from the qualified
archive path (e.g. use a mock FS with a scheme and authority) so it fails
without `makeQualified`?
--
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]
