sarvekshayr commented on code in PR #11097:
URL: https://github.com/apache/ozone/pull/11097#discussion_r4245506989
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportFileManager.java:
##########
@@ -117,96 +129,66 @@ private void lock() throws IOException {
}
}
- File resolveArchiveFile(ExportScope scope, String jobStartTime, ExportJob.Id
jobId) {
- return new File(exportDirectory, String.format("container-ids_%s_%s%s%s%s",
- scope.getValue(), jobStartTime, EXPORT_ARCHIVE_JOB_INFIX,
jobId.getValue(), EXPORT_ARCHIVE_SUFFIX));
+ void unlock() throws IOException {
+ if (exportDirectoryLock == null) {
+ return;
+ }
+ exportDirectoryLock.release();
+ exportDirectoryLock.channel().close();
+ exportDirectoryLock = null;
}
- File resolveArchiveTempFile(ExportScope scope, String jobStartTime,
ExportJob.Id jobId) {
- return AtomicFileOutputStream.getTemporaryFile(resolveArchiveFile(scope,
jobStartTime, jobId));
+ static String archiveFileName(ExportScope scope, String jobStartTime,
ExportJob.Id jobId) {
+ return String.format("container-ids_%s_%s%s%s%s",
+ scope, jobStartTime, EXPORT_ARCHIVE_JOB_INFIX, jobId,
EXPORT_ARCHIVE_SUFFIX);
}
void createJobDirectory(ExportJob.Id jobId) throws IOException {
Files.createDirectories(jobDirectory(jobId));
}
- BufferedWriter newPartWriter(ExportJob.Id jobId, String partFileName) throws
IOException {
- return Files.newBufferedWriter(jobDirectory(jobId).resolve(partFileName),
StandardCharsets.UTF_8);
+ OutputStream newPartOutputStream(ExportJob.Id jobId, String partFileName)
throws IOException {
+ return Files.newOutputStream(jobDirectory(jobId).resolve(partFileName));
}
- void writeArchive(ExportJob.Id jobId, String archivePath) throws IOException
{
- Path jobDir = jobDirectory(jobId);
+ void writeArchive(ExportJob job, BooleanSupplier isLeaderReady) throws
IOException {
+ Path jobDir = jobDirectory(job.getId());
File[] parts = jobDir.toFile().listFiles((dir, name) ->
name.endsWith(".txt"));
if (parts == null || parts.length == 0) {
- throw new IOException("No part files found for export job " + jobId);
+ throw new IOException("No part files found for " + job);
}
Arrays.sort(parts, Comparator.comparing(File::getName));
- File archiveFile = new File(archivePath);
+ File archiveFile = new File(exportDirectory, job.getFileName());
try (AtomicFileOutputStream atomicOut = new
AtomicFileOutputStream(archiveFile);
GZIPOutputStream gzipOut = new GZIPOutputStream(atomicOut);
ArchiveOutputStream<TarArchiveEntry> tarOut = Archiver.tar(gzipOut)) {
for (File part : parts) {
+ if (!isLeaderReady.getAsBoolean()) {
+ // close() would publish the partial archive; abort() discards the
temp file instead.
+ atomicOut.abort();
+ throw new IOException("SCM lost leadership while archiving " + job);
+ }
Archiver.includeFile(part, part.getName(), tarOut);
}
}
Review Comment:
Archiving is the last step of a long-running job. Failing it because
leadership flipped in the final few milliseconds throws away a correct archive.
Keeping it costs the operator at most one extra archive from the old leader.
The contents may be slightly stale, which the class javadoc already says
(best-effort, not point-in-time).
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
