spuru9 commented on code in PR #29008:
URL: https://github.com/apache/flink/pull/29008#discussion_r4497253125
##########
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:
major: Opening an uncached old job with parallel requests, which is what the
dashboard does, mostly 404s. The first request reads the archive synchronously
before it submits the detail task, and the others arrive meanwhile, see
OVERVIEW_PARSING, skip the fetch, find no task in
`waitLazyFetchArchiveFinished` and return straight away. Most of the parallel
requests fail for me on both FILE and RocksDB; on FILE it only worked when the
eviction delete of a big job happened to hold the lock long enough.
##########
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:
major: If the on-demand read throws once (a truncated archive or a flaky S3
read does it), the meta entry stays in OVERVIEW_PARSING.
`needLazyLoadIndividually` then never refetches, so the job keeps returning 404
even after the archive is readable again, and this check treats it as in flight
forever, so it's never evicted either. The root cause is older, but this PR
makes on-demand fetch the normal path for every old job, so the meta entry
should be cleared on failure.
--
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]
