lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-4332] The current instant may be wrong under some extreme conditions in AppendWriteFunction. (#5988)

Browse Source
This commit is contained in:
BruceLin
2022-06-28 20:42:26 +08:00
committed by GitHub
parent 08eba914ed
commit efb9719018
1 changed files with 7 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunction.java
View File

@@ -106,30 +106,29 @@ public class AppendWriteFunction<I> extends AbstractStreamWriteFunction<I> {
// Utilities
// -------------------------------------------------------------------------
private void initWriterHelper() {
this.currentInstant = instantToWrite(true);
if (this.currentInstant == null) {
final String instant = instantToWrite(true);
if (instant == null) {
// in case there are empty checkpoints that has no input data
throw new HoodieException("No inflight instant when flushing data!");
}
this.writerHelper = new BulkInsertWriterHelper(this.config, this.writeClient.getHoodieTable(), this.writeClient.getConfig(),
this.currentInstant, this.taskID, getRuntimeContext().getNumberOfParallelSubtasks(), getRuntimeContext().getAttemptNumber(),
instant, this.taskID, getRuntimeContext().getNumberOfParallelSubtasks(), getRuntimeContext().getAttemptNumber(),
this.rowType);
}
private void flushData(boolean endInput) {
final List<WriteStatus> writeStatus;
final String instant;
if (this.writerHelper != null) {
writeStatus = this.writerHelper.getWriteStatuses(this.taskID);
instant = this.writerHelper.getInstantTime();
this.currentInstant = this.writerHelper.getInstantTime();
} else {
writeStatus = Collections.emptyList();
instant = instantToWrite(false);
LOG.info("No data to write in subtask [{}] for instant [{}]", taskID, instant);
this.currentInstant = instantToWrite(false);
LOG.info("No data to write in subtask [{}] for instant [{}]", taskID, this.currentInstant);
}
final WriteMetadataEvent event = WriteMetadataEvent.builder()
.taskID(taskID)
.instantTime(instant)
.instantTime(this.currentInstant)
.writeStatus(writeStatus)
.lastBatch(true)
.endInput(endInput)