lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

HUDI-168 Ensure getFileStatus calls for files getting written is done after close() is called (#788)

Browse Source
This commit is contained in:
Balaji Varadarajan
2019-07-16 17:33:34 -07:00
committed by vinoth chandar
parent c0593e7a13
commit 3d408ee96b
1 changed files with 9 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
hoodie-client/src/main/java/com/uber/hoodie/table/HoodieMergeOnReadTable.java
View File

@@ -505,20 +505,18 @@ public class HoodieMergeOnReadTable<T extends HoodieRecordPayload> extends
}).forEach(wStat -> {
HoodieLogFormat.Writer writer = null;
String baseCommitTime = fileIdToBaseCommitTimeForLogMap.get(wStat.getFileId());
boolean success = false;
try {
writer = HoodieLogFormat.newWriterBuilder().onParentPath(
FSUtils.getPartitionPath(this.getMetaClient().getBasePath(), partitionPath))
.withFileId(wStat.getFileId()).overBaseCommit(baseCommitTime)
.withFs(this.metaClient.getFs())
.withFileExtension(HoodieLogFile.DELTA_EXTENSION).build();
Long numRollbackBlocks = 0L;
// generate metadata
Map<HeaderMetadataType, String> header = generateHeader(commit);
// if update belongs to an existing log file
writer = writer.appendBlock(new HoodieCommandBlock(header));
numRollbackBlocks++;
filesToNumBlocksRollback.put(this.getMetaClient().getFs()
.getFileStatus(writer.getLogFile().getPath()), numRollbackBlocks);
success = true;
} catch (IOException | InterruptedException io) {
throw new HoodieRollbackException(
"Failed to rollback for commit " + commit, io);
@@ -527,6 +525,13 @@ public class HoodieMergeOnReadTable<T extends HoodieRecordPayload> extends
if (writer != null) {
writer.close();
}
if (success) {
// This step is intentionally done after writer is closed. Guarantees that
// getFileStatus would reflect correct stats and FileNotFoundException is not thrown in
// cloud-storage : HUDI-168
filesToNumBlocksRollback.put(this.getMetaClient().getFs()
.getFileStatus(writer.getLogFile().getPath()), 1L);
}
} catch (IOException io) {
throw new UncheckedIOException(io);
}