Rename staging file name with record count before moving to task output
Closes from vikrambohra/GOBBLIN-1411
@@ -70,7 +70,7 @@ public abstract class FsDataWriter<D> implements DataWriter<D>, FinalState, Meta
protected final String fileName;
protected final FileSystem fs;
protected final FileContext fileContext;
- protected final Path stagingFile;
+ protected Path stagingFile;
protected final String partitionKey;
private final GlobalMetadata defaultMetadata;
protected Path outputFile;
@@ -256,18 +256,18 @@ public abstract class FsDataWriter<D> implements DataWriter<D>, FinalState, Meta
this.bytesWritten = Optional.of(Long.valueOf(stagingFileStatus.getLen()));
+ // Rename staging file to add record count before copying to output file
+ if (this.shouldIncludeRecordCountInFileName) {
+ String filePathWithRecordCount = addRecordCountToStagingFile();
+ this.stagingFile = new Path(filePathWithRecordCount);
+ this.outputFile = new Path(this.outputFile.getParent().toString(), new Path(filePathWithRecordCount).getName());
+ }
+
LOG.info(String.format("Moving data from %s to %s", this.stagingFile, this.outputFile));
// For the same reason as deleting the staging file if it already exists, overwrite
// the output file if it already exists to prevent task retry from being blocked.
HadoopUtils.renamePath(this.fs, this.stagingFile, this.outputFile, true);
-
- // The staging file is moved to the output path in commit, so rename to add record count after that
- if (this.shouldIncludeRecordCountInFileName) {
- String filePathWithRecordCount = addRecordCountToFileName();
- this.properties.appendToSetProp(this.allOutputFilesPropName, filePathWithRecordCount);
- } else {
- this.properties.appendToSetProp(this.allOutputFilesPropName, getOutputFilePath());
- }
+ this.properties.appendToSetProp(this.allOutputFilesPropName, this.outputFile.toString());
FsWriterMetrics metrics = new FsWriterMetrics(
this.id,
@@ -301,13 +301,12 @@ public abstract class FsDataWriter<D> implements DataWriter<D>, FinalState, Meta
this.closer.close();
}
- private synchronized String addRecordCountToFileName()
+ private synchronized String addRecordCountToStagingFile()
throws IOException {
- String filePath = getOutputFilePath();
+ String filePath = this.stagingFile.toString();
String filePathWithRecordCount = IngestionRecordCountProvider.constructFilePath(filePath, recordsWritten());
LOG.info("Renaming " + filePath + " to " + filePathWithRecordCount);
HadoopUtils.renamePath(this.fs, new Path(filePath), new Path(filePathWithRecordCount), true);
- this.outputFile = new Path(filePathWithRecordCount);
return filePathWithRecordCount;
}
