Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
danny0405 merged 9 commits into from
Apr 20, 2022
Merged

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

Changes from 8 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
cb1112f
fix lose data when compact rollback
wxplovecc Apr 19, 2022
771376d
fix lose data when compact rollback
wxplovecc Apr 19, 2022
d9c0a04
stop add event when has failed compact event
wxplovecc Apr 19, 2022
7de9f66
stop add event when has failed compact event
wxplovecc Apr 19, 2022
611fd11
stop add event when has failed compact event
wxplovecc Apr 19, 2022
b9bfb0a
format code
wxplovecc Apr 19, 2022
d71567e
format code
wxplovecc Apr 19, 2022
3352b6d
format code
Apr 20, 2022
a312444
change comment
Apr 20, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 9 additions & 5 deletions ...atasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,11 +101,6 @@ public void open(Configuration parameters) throws Exception {
@Override
public void invoke(CompactionCommitEvent event, Context context) throws Exception {
final String instant = event.getInstant();
if (event.isFailed()) {
// handle failure case
CompactionUtil.rollbackCompaction(table, event.getInstant());
return;
}
commitBuffer.computeIfAbsent(instant, k -> new HashMap<>())
.put(event.getFileId(), event);
commitIfNecessary(instant, commitBuffer.get(instant).values());
Expand All @@ -132,6 +127,15 @@ private void commitIfNecessary(String instant, Collection<CompactionCommitEvent>
if (!isReady) {
return;
}

if (events.stream().anyMatch(CompactionCommitEvent::isFailed)) {
// handle failure case
CompactionUtil.rollbackCompaction(table, instant);
// remove commitBuffer avoid commit with preview fileId
reset(instant);
return;
}

Copy link
Contributor

@danny0405 danny0405 Apr 20, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we ensure the buffer was cleared like this:

try {
  // handle failure case
  CompactionUtil.rollbackCompaction(table, instant);
} finally {
  // remove commitBuffer to avoid obsolete metadata commit
  reset(instant);
  return;
}

try {
doCommit(instant, events);
} catch (Throwable throwable) {
Expand Down