You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2021/10/18 07:43:15 UTC
[flink] 02/04: [hotfix] Refactor JobArchiveFetcherTask#run
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f182a96041c71c7c640399f94d11ce968676974d
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Mon Oct 4 13:50:09 2021 +0200
[hotfix] Refactor JobArchiveFetcherTask#run
---
.../history/HistoryServerArchiveFetcher.java | 131 ++++++++++++---------
1 file changed, 74 insertions(+), 57 deletions(-)
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java
index c9cd0a4..a547aff 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java
@@ -217,13 +217,12 @@ class HistoryServerArchiveFetcher {
Map<Path, Set<Path>> archivesBeyondSizeLimit = new HashMap<>();
for (HistoryServer.RefreshLocation refreshLocation : refreshDirs) {
Path refreshDir = refreshLocation.getPath();
- FileSystem refreshFS = refreshLocation.getFs();
LOG.debug("Checking archive directory {}.", refreshDir);
// contents of /:refreshDir
FileStatus[] jobArchives;
try {
- jobArchives = refreshFS.listStatus(refreshDir);
+ jobArchives = listArchives(refreshLocation.getFs(), refreshDir);
} catch (IOException e) {
LOG.error(
"Failed to access job archive location for path {}.",
@@ -234,27 +233,12 @@ class HistoryServerArchiveFetcher {
jobsToRemove.remove(refreshDir);
continue;
}
- if (jobArchives == null) {
- // the entire refreshDirectory was removed
- continue;
- }
-
- Arrays.sort(
- jobArchives,
- Comparator.comparingLong(FileStatus::getModificationTime).reversed());
int historySize = 0;
for (FileStatus jobArchive : jobArchives) {
Path jobArchivePath = jobArchive.getPath();
String jobID = jobArchivePath.getName();
- try {
- JobID.fromHexString(jobID);
- } catch (IllegalArgumentException iae) {
- LOG.debug(
- "Archive directory {} contained file with unexpected name {}. Ignoring file.",
- refreshDir,
- jobID,
- iae);
+ if (!isValidJobID(jobID, refreshDir)) {
continue;
}
@@ -275,45 +259,7 @@ class HistoryServerArchiveFetcher {
} else {
LOG.info("Processing archive {}.", jobArchivePath);
try {
- for (ArchivedJson archive :
- FsJobArchivist.getArchivedJsons(jobArchive.getPath())) {
- String path = archive.getPath();
- String json = archive.getJson();
-
- File target;
- if (path.equals(JobsOverviewHeaders.URL)) {
- target = new File(webOverviewDir, jobID + JSON_FILE_ENDING);
- } else if (path.equals("/joboverview")) { // legacy path
- LOG.debug("Migrating legacy archive {}", jobArchivePath);
- json = convertLegacyJobOverview(json);
- target = new File(webOverviewDir, jobID + JSON_FILE_ENDING);
- } else {
- // this implicitly writes into webJobDir
- target = new File(webDir, path + JSON_FILE_ENDING);
- }
-
- java.nio.file.Path parent = target.getParentFile().toPath();
-
- try {
- Files.createDirectories(parent);
- } catch (FileAlreadyExistsException ignored) {
- // there may be left-over directories from the previous
- // attempt
- }
-
- java.nio.file.Path targetPath = target.toPath();
-
- // We overwrite existing files since this may be another attempt
- // at fetching this archive.
- // Existing files may be incomplete/corrupt.
- Files.deleteIfExists(targetPath);
-
- Files.createFile(target.toPath());
- try (FileWriter fw = new FileWriter(target)) {
- fw.write(json);
- fw.flush();
- }
- }
+ processArchive(jobID, jobArchivePath);
events.add(new ArchiveEvent(jobID, ArchiveEventType.CREATED));
cachedArchivesPerRefreshDirectory.get(refreshDir).add(jobID);
LOG.info("Processing archive {} finished.", jobArchivePath);
@@ -345,6 +291,77 @@ class HistoryServerArchiveFetcher {
}
}
+ private static FileStatus[] listArchives(FileSystem refreshFS, Path refreshDir)
+ throws IOException {
+ // contents of /:refreshDir
+ FileStatus[] jobArchives = refreshFS.listStatus(refreshDir);
+ if (jobArchives == null) {
+ // the entire refreshDirectory was removed
+ return new FileStatus[0];
+ }
+
+ Arrays.sort(
+ jobArchives,
+ Comparator.comparingLong(FileStatus::getModificationTime).reversed());
+
+ return jobArchives;
+ }
+
+ private static boolean isValidJobID(String jobId, Path refreshDir) {
+ try {
+ JobID.fromHexString(jobId);
+ return true;
+ } catch (IllegalArgumentException iae) {
+ LOG.debug(
+ "Archive directory {} contained file with unexpected name {}. Ignoring file.",
+ refreshDir,
+ jobId,
+ iae);
+ return false;
+ }
+ }
+
+ private void processArchive(String jobID, Path jobArchive) throws IOException {
+ for (ArchivedJson archive : FsJobArchivist.getArchivedJsons(jobArchive)) {
+ String path = archive.getPath();
+ String json = archive.getJson();
+
+ File target;
+ if (path.equals(JobsOverviewHeaders.URL)) {
+ target = new File(webOverviewDir, jobID + JSON_FILE_ENDING);
+ } else if (path.equals("/joboverview")) { // legacy path
+ LOG.debug("Migrating legacy archive {}", jobArchive);
+ json = convertLegacyJobOverview(json);
+ target = new File(webOverviewDir, jobID + JSON_FILE_ENDING);
+ } else {
+ // this implicitly writes into webJobDir
+ target = new File(webDir, path + JSON_FILE_ENDING);
+ }
+
+ java.nio.file.Path parent = target.getParentFile().toPath();
+
+ try {
+ Files.createDirectories(parent);
+ } catch (FileAlreadyExistsException ignored) {
+ // there may be left-over directories from the previous
+ // attempt
+ }
+
+ java.nio.file.Path targetPath = target.toPath();
+
+ // We overwrite existing files since this may be another attempt
+ // at fetching this archive.
+ // Existing files may be incomplete/corrupt.
+ Files.deleteIfExists(targetPath);
+
+ Files.createFile(target.toPath());
+ try (FileWriter fw = new FileWriter(target)) {
+ fw.write(json);
+ fw.flush();
+ }
+ }
+ }
+
private List<ArchiveEvent> cleanupJobsBeyondSizeLimit(
Map<Path, Set<Path>> jobArchivesToRemove) {
Map<Path, Set<String>> allJobIdsToRemoveFromOverview = new HashMap<>();