lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-3870] Add timeout rollback for flink online compaction (#5314)

Browse Source
This commit is contained in:
Danny Chan
2022-04-13 20:05:48 +08:00
committed by GitHub
parent 0281725c6b
commit 6f9b02decb
1 changed files with 10 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionPlanOperator.java
View File

@@ -88,8 +88,7 @@ public class CompactionPlanOperator extends AbstractStreamOperator<CompactionPla
// when the earliest inflight instant has timed out, assumes it has failed // when the earliest inflight instant has timed out, assumes it has failed
// already and just rolls it back. // already and just rolls it back.
// comment out: do we really need the timeout rollback ? CompactionUtil.rollbackEarliestCompaction(table, conf);
// CompactionUtil.rollbackEarliestCompaction(table, conf);
scheduleCompaction(table, checkpointId); scheduleCompaction(table, checkpointId);
} catch (Throwable throwable) { } catch (Throwable throwable) {
// make it fail-safe // make it fail-safe
@@ -99,7 +98,8 @@ public class CompactionPlanOperator extends AbstractStreamOperator<CompactionPla
private void scheduleCompaction(HoodieFlinkTable<?> table, long checkpointId) throws IOException { private void scheduleCompaction(HoodieFlinkTable<?> table, long checkpointId) throws IOException {
// the first instant takes the highest priority. // the first instant takes the highest priority.
Option<HoodieInstant> firstRequested = table.getActiveTimeline().filterPendingCompactionTimeline() HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
Option<HoodieInstant> firstRequested = pendingCompactionTimeline
.filter(instant -> instant.getState() == HoodieInstant.State.REQUESTED).firstInstant(); .filter(instant -> instant.getState() == HoodieInstant.State.REQUESTED).firstInstant();
if (!firstRequested.isPresent()) { if (!firstRequested.isPresent()) {
// do nothing. // do nothing.
@@ -107,6 +107,13 @@ public class CompactionPlanOperator extends AbstractStreamOperator<CompactionPla
return; return;
} }
Option<HoodieInstant> firstInflight = pendingCompactionTimeline
.filter(instant -> instant.getState() == HoodieInstant.State.INFLIGHT).firstInstant();
if (firstInflight.isPresent()) {
LOG.warn("Waiting for pending compaction instant : " + firstInflight + " to complete, skip scheduling new compaction plans");
return;
}
String compactionInstantTime = firstRequested.get().getTimestamp(); String compactionInstantTime = firstRequested.get().getTimestamp();
// generate compaction plan // generate compaction plan