diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java
index 4728f717d..b08194b8d 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java
@@ -226,17 +226,14 @@ public class HoodieFlinkCompactor {
 
       // checks the compaction plan and do compaction.
       if (cfg.schedule) {
-        Option<String> compactionInstantTimeOption = CompactionUtil.getCompactionInstantTime(metaClient);
-        if (compactionInstantTimeOption.isPresent()) {
-          boolean scheduled = writeClient.scheduleCompactionAtInstant(compactionInstantTimeOption.get(), Option.empty());
-          if (!scheduled) {
-            // do nothing.
-            LOG.info("No compaction plan for this job ");
-            compactClosed("No compaction plan for this job ", null);
-            return;
-          }
-          table.getMetaClient().reloadActiveTimeline();
+        boolean scheduled = writeClient.scheduleCompaction(Option.empty()).isPresent();
+        if (!scheduled) {
+          // do nothing.
+          LOG.info("No compaction plan for this job ");
+          compactClosed("No compaction plan for this job ", null);
+          return;
         }
+        table.getMetaClient().reloadActiveTimeline();
       }
 
       // fetch the instant based on the configured execution sequence