[HUDI-3870] Add timeout rollback for flink online compaction (apache#…

…5314)
uday08bce · Apr 13, 2022 · 6f9b02d · 6f9b02d
1 parent 0281725
commit 6f9b02d
Showing 1 changed file with 10 additions and 3 deletions.
diff --git a/...asource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionPlanOperator.java b/...asource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionPlanOperator.java
@@ -88,8 +88,7 @@ public void notifyCheckpointComplete(long checkpointId) {
       // when the earliest inflight instant has timed out, assumes it has failed
       // 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);
     } catch (Throwable throwable) {
       // make it fail-safe
@@ -99,14 +98,22 @@ public void notifyCheckpointComplete(long checkpointId) {
 
   private void scheduleCompaction(HoodieFlinkTable<?> table, long checkpointId) throws IOException {
     // 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();
     if (!firstRequested.isPresent()) {
       // do nothing.
       LOG.info("No compaction plan for checkpoint " + checkpointId);
       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();
 
     // generate compaction plan