argoyal2212 commented on code in PR #29008:
URL: https://github.com/apache/flink/pull/29008#discussion_r4500314578
##########
flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandlerTest.java:
##########
@@ -374,4 +376,59 @@ void testLazyModeLoadResource() throws Exception {
assertThat(detailResponse.get().f1).contains(jobId.toString());
assertThat(blockingArchiveStorage.exists("jobs/" + jobId +
"/config.json")).isTrue();
}
+
+ /**
+ * Two users open two jobs outside the cached window in {@code LAZY} mode.
Viewing the second
+ * job must not cancel the first job's fetch that is still in flight.
+ */
+ @TestTemplate
+ void testLazyModeViewingAnotherJobDoesNotCancelInFlightFetch() throws
Exception {
+ startServer(LAZY);
+
+ JobID newest = JobID.generate();
+ JobID first = JobID.generate();
+ JobID second = JobID.generate();
+ long now = System.currentTimeMillis();
+ long[] times = {now - 3000, now - 2000, now - 1000};
+ JobID[] ids = {first, second, newest};
+ for (int i = 0; i < ids.length; i++) {
+ org.apache.flink.core.fs.Path archive =
+ createJobArchive(remoteArchiveRootPath.toFile(), ids[i],
true);
+ assertThat(new
File(archive.toUri()).setLastModified(times[i])).isTrue();
+ }
+
+ assertThat(HttpUtils.getFromHTTP(baseUrl +
"/jobs/overview").f0).isEqualTo(200);
+
+ HistoryServerTestUtils.BlockingArchiveStorage<?> storage =
+ (HistoryServerTestUtils.BlockingArchiveStorage<?>)
handler.archiveStorage;
+
+ AtomicReference<Tuple2<Integer, String>> firstResponse = new
AtomicReference<>();
+ CompletableFuture<Void> firstFuture = requestAsync(first,
firstResponse);
+ Thread.sleep(500);
+ assertThat(firstFuture.isDone()).isFalse();
+
+ // viewing another old job while the first fetch is still blocked
+ AtomicReference<Tuple2<Integer, String>> secondResponse = new
AtomicReference<>();
+ CompletableFuture<Void> secondFuture = requestAsync(second,
secondResponse);
+ Thread.sleep(500);
Review Comment:
Good point, replaced both sleeps with polling on the actual storage state.
##########
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:
Agreed, the test now views two old jobs and asserts the first one gets
evicted, and it fails without the fix.
##########
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:
Fixed, in-flight is now decided from the meta event type so a finished fetch
no longer counts, and I added a test for it.
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandler.java:
##########
@@ -347,6 +348,9 @@ private Entry loadResource(
jobArchiveMetaInfo == null ? null :
jobArchiveMetaInfo.getArchivePath();
archiveFetcher.lazyFetchArchiveProactively(jobId,
archivePath);
}
+ if (archiveFetcher.getArchiveMetaInfo(jobId) != null) {
Review Comment:
Fixed, extractJobId now strips the .json suffix so /jobs/<id> triggers the
lazy fetch, with a test for it.
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandler.java:
##########
@@ -347,6 +348,9 @@ private Entry loadResource(
jobArchiveMetaInfo == null ? null :
jobArchiveMetaInfo.getArchivePath();
archiveFetcher.lazyFetchArchiveProactively(jobId,
archivePath);
}
+ if (archiveFetcher.getArchiveMetaInfo(jobId) != null) {
+ archiveFetcher.markRecentlyViewed(jobId);
Review Comment:
Fixed, the fetch future is registered before the sync read so parallel
requests wait on it, and I checked it with a real history server under parallel
load.
##########
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)) {
+ return true;
+ }
+ ArchiveMetaInfo metaInfo = archiveMetaInfoCache.get(jobId);
+ if (metaInfo == null) {
+ return false;
+ }
+ switch (metaInfo.getEventType()) {
Review Comment:
Fixed, a failed on-demand read now clears the meta and the future so the
next request retries, covered by a new test.
--
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]
