lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

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

Browse Source 
This reverts commit 6f9b02decb.
This commit is contained in:
Danny Chan
2022-05-18 20:30:54 +08:00
committed by GitHub
parent 199f64255e
commit 551aa959c5
1 changed files with 3 additions and 10 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

@@ -89,7 +89,8 @@ 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.
CompactionUtil.rollbackEarliestCompaction(table, conf); // comment out: do we really need the timeout rollback ?
// 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,8 +100,7 @@ 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.
HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline(); Option<HoodieInstant> firstRequested = 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.
@@ -108,13 +108,6 @@ 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