lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-3912] Fix lose data when rollback in flink async compact (#5357)

Browse Source 
* stop add event when has failed compact event

Co-authored-by: wxp <wxp4532@outlook.com>
This commit is contained in:
吴祥平
2022-04-20 19:23:39 +08:00
committed by GitHub
parent 6a3ce928b1
commit 408663c42b
1 changed files with 12 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitSink.java
View File

@@ -101,11 +101,6 @@ public class CompactionCommitSink extends CleanFunction<CompactionCommitEvent> {
@Override @Override
public void invoke(CompactionCommitEvent event, Context context) throws Exception { public void invoke(CompactionCommitEvent event, Context context) throws Exception {
final String instant = event.getInstant(); final String instant = event.getInstant();
if (event.isFailed()) {
// handle failure case
CompactionUtil.rollbackCompaction(table, event.getInstant());
return;
}
commitBuffer.computeIfAbsent(instant, k -> new HashMap<>()) commitBuffer.computeIfAbsent(instant, k -> new HashMap<>())
.put(event.getFileId(), event); .put(event.getFileId(), event);
commitIfNecessary(instant, commitBuffer.get(instant).values()); commitIfNecessary(instant, commitBuffer.get(instant).values());
@@ -132,6 +127,18 @@ public class CompactionCommitSink extends CleanFunction<CompactionCommitEvent> {
if (!isReady) { if (!isReady) {
return; return;
} }
if (events.stream().anyMatch(CompactionCommitEvent::isFailed)) {
try {
// handle failure case
CompactionUtil.rollbackCompaction(table, instant);
} finally {
// remove commitBuffer to avoid obsolete metadata commit
reset(instant);
return;
}
}
try { try {
doCommit(instant, events); doCommit(instant, events);
} catch (Throwable throwable) { } catch (Throwable throwable) {