From 551aa959c57721a5cc4d3f63f79e0201978980a2 Mon Sep 17 00:00:00 2001
From: Danny Chan <yuzhao.cyz@gmail.com>
Date: Wed, 18 May 2022 20:30:54 +0800
Subject: [PATCH] Revert "[HUDI-3870] Add timeout rollback for flink online
 compaction (#5314)" (#5622)

This reverts commit 6f9b02decb5bb2b83709b1b6ec04a97e4d102c11.
---
 .../hudi/sink/compact/CompactionPlanOperator.java   | 13 +++----------
 1 file changed, 3 insertions(+), 10 deletions(-)

diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionPlanOperator.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionPlanOperator.java
index 338352d4b..d5e718883 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionPlanOperator.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionPlanOperator.java
@@ -89,7 +89,8 @@ public class CompactionPlanOperator extends AbstractStreamOperator<CompactionPla
       // when the earliest inflight instant has timed out, assumes it has failed
       // 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);
     } catch (Throwable throwable) {
       // make it fail-safe
@@ -99,8 +100,7 @@ public class CompactionPlanOperator extends AbstractStreamOperator<CompactionPla
 
   private void scheduleCompaction(HoodieFlinkTable<?> table, long checkpointId) throws IOException {
     // the first instant takes the highest priority.
-    HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
-    Option<HoodieInstant> firstRequested = pendingCompactionTimeline
+    Option<HoodieInstant> firstRequested = table.getActiveTimeline().filterPendingCompactionTimeline()
         .filter(instant -> instant.getState() == HoodieInstant.State.REQUESTED).firstInstant();
     if (!firstRequested.isPresent()) {
       // do nothing.
@@ -108,13 +108,6 @@ public class CompactionPlanOperator extends AbstractStreamOperator<CompactionPla
       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