[HUDI-2040] Make flink writer as exactly-once by default (#3106)

hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java

@@ -315,14 +315,6 @@ public class FlinkOptions {
       .defaultValue(100) // default 100 MB
       .withDescription("Max memory in MB for merge, default 100MB");
 
-  public static final ConfigOption<Boolean> WRITE_EXACTLY_ONCE_ENABLED = ConfigOptions
-      .key("write.exactly_once.enabled")
-      .booleanType()
-      .defaultValue(false) // default at least once
-      .withDescription("Whether write in exactly_once semantics, if true,\n"
-          + "the write task would block flushing after it finishes a checkpoint\n"
-          + "until it receives the checkpoint success event, default false");
-
   // this is only for internal use
   public static final ConfigOption<Long> WRITE_COMMIT_ACK_TIMEOUT = ConfigOptions
       .key("write.commit.ack.timeout")

hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java

@@ -37,7 +37,6 @@ import org.apache.hudi.table.action.commit.FlinkWriteHelper;
 import org.apache.hudi.util.StreamerUtil;
 
 import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.api.common.state.CheckpointListener;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.operators.coordination.OperatorEventGateway;
 import org.apache.flink.runtime.state.FunctionInitializationContext;
@@ -100,7 +99,7 @@ import java.util.stream.Collectors;
  */
 public class StreamWriteFunction<K, I, O>
     extends KeyedProcessFunction<K, I, O>
-    implements CheckpointedFunction, CheckpointListener {
+    implements CheckpointedFunction {
 
   private static final long serialVersionUID = 1L;
 
@@ -148,11 +147,6 @@ public class StreamWriteFunction<K, I, O>
    */
   private transient TotalSizeTracer tracer;
 
-  /**
-   * Whether write in exactly-once semantics.
-   */
-  private boolean exactlyOnce;
-
   /**
    * Flag saying whether the write task is waiting for the checkpoint success notification
    * after it finished a checkpoint.
@@ -186,7 +180,6 @@ public class StreamWriteFunction<K, I, O>
         WriteOperationType.fromValue(config.getString(FlinkOptions.OPERATION)),
         HoodieTableType.valueOf(config.getString(FlinkOptions.TABLE_TYPE)));
     this.tracer = new TotalSizeTracer(this.config);
-    this.exactlyOnce = config.getBoolean(FlinkOptions.WRITE_EXACTLY_ONCE_ENABLED);
     initBuffer();
     initWriteFunction();
   }
@@ -217,11 +210,6 @@ public class StreamWriteFunction<K, I, O>
     }
   }
 
-  @Override
-  public void notifyCheckpointComplete(long checkpointId) {
-    this.writeClient.cleanHandles();
-  }
-
   /**
    * End input action for batch source.
    */
@@ -502,7 +490,7 @@ public class StreamWriteFunction<K, I, O>
 
     // if exactly-once semantics turns on,
     // waits for the checkpoint notification until the checkpoint timeout threshold hits.
-    if (exactlyOnce && confirming) {
+    if (confirming) {
       long waitingTime = 0L;
       long ckpTimeout = config.getLong(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT);
       long interval = 500L;
@@ -583,6 +571,7 @@ public class StreamWriteFunction<K, I, O>
     this.eventGateway.sendEventToCoordinator(event);
     this.buckets.clear();
     this.tracer.reset();
+    this.writeClient.cleanHandles();
     this.confirming = true;
   }
 }

hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java

@@ -104,6 +104,11 @@ public class StreamWriteOperatorCoordinator
    */
   private final int parallelism;
 
+  /**
+   * Whether to schedule asynchronous compaction task on finished checkpoints.
+   */
+  private final boolean asyncCompaction;
+
   /**
    * A single-thread executor to handle all the asynchronous jobs of the coordinator.
    */
@@ -136,6 +141,7 @@ public class StreamWriteOperatorCoordinator
     this.conf = conf;
     this.context = context;
     this.parallelism = context.currentParallelism();
+    this.asyncCompaction = StreamerUtil.needsAsyncCompaction(conf);
   }
 
   @Override
@@ -197,6 +203,10 @@ public class StreamWriteOperatorCoordinator
           // so a successful checkpoint subsumes the old one(follows the checkpoint subsuming contract)
           final boolean committed = commitInstant();
           if (committed) {
+            // if async compaction is on, schedule the compaction
+            if (asyncCompaction) {
+              writeClient.scheduleCompaction(Option.empty());
+            }
             // start new instant.
             startInstant();
             // sync Hive if is enabled

hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionPlanOperator.java

@@ -21,13 +21,11 @@ package org.apache.hudi.sink.compact;
 import org.apache.hudi.avro.model.HoodieCompactionPlan;
 import org.apache.hudi.client.HoodieFlinkWriteClient;
 import org.apache.hudi.common.model.CompactionOperation;
-import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
 import org.apache.hudi.common.util.CompactionUtils;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.table.HoodieFlinkTable;
-import org.apache.hudi.util.CompactionUtil;
 import org.apache.hudi.util.StreamerUtil;
 
 import org.apache.flink.annotation.VisibleForTesting;
@@ -92,17 +90,18 @@ public class CompactionPlanOperator extends AbstractStreamOperator<CompactionPla
   }
 
   private void scheduleCompaction(long checkpointId) throws IOException {
-    // if async compaction is on, schedule the compaction
-    HoodieTableMetaClient metaClient = CompactionUtil.createMetaClient(conf);
-    final String compactionInstantTime = CompactionUtil.getCompactionInstantTime(metaClient);
+    HoodieFlinkTable<?> table = writeClient.getHoodieTable();
 
-    boolean scheduled = writeClient.scheduleCompactionAtInstant(compactionInstantTime, Option.empty());
-    if (!scheduled) {
+    // the last instant takes the highest priority.
+    Option<HoodieInstant> lastRequested = table.getActiveTimeline().filterPendingCompactionTimeline()
+        .filter(instant -> instant.getState() == HoodieInstant.State.REQUESTED).lastInstant();
+    if (!lastRequested.isPresent()) {
       // do nothing.
       LOG.info("No compaction plan for checkpoint " + checkpointId);
       return;
     }
 
+    String compactionInstantTime = lastRequested.get().getTimestamp();
     if (this.compactionInstantTime != null
         && Objects.equals(this.compactionInstantTime, compactionInstantTime)) {
       // do nothing
@@ -110,7 +109,6 @@ public class CompactionPlanOperator extends AbstractStreamOperator<CompactionPla
       return;
     }
 
-    HoodieFlinkTable<?> table = writeClient.getHoodieTable();
     // generate compaction plan
     // should support configurable commit metadata
     HoodieCompactionPlan compactionPlan = CompactionUtils.getCompactionPlan(
@@ -123,27 +121,13 @@ public class CompactionPlanOperator extends AbstractStreamOperator<CompactionPla
     } else {
       this.compactionInstantTime = compactionInstantTime;
       HoodieInstant instant = HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
-      HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
-      if (!pendingCompactionTimeline.containsInstant(instant)) {
-        // this means that the compaction plan was written to auxiliary path(.tmp)
-        // but not the meta path(.hoodie), this usually happens when the job crush
-        // exceptionally.
-
-        // clean the compaction plan in auxiliary path and cancels the compaction.
-
-        LOG.warn("The compaction plan was fetched through the auxiliary path(.tmp) but not the meta path(.hoodie).\n"
-            + "Clean the compaction plan in auxiliary path and cancels the compaction");
-        CompactionUtil.cleanInstant(table.getMetaClient(), instant);
-        return;
-      }
-
       // Mark instant as compaction inflight
       table.getActiveTimeline().transitionCompactionRequestedToInflight(instant);
       table.getMetaClient().reloadActiveTimeline();
 
       List<CompactionOperation> operations = compactionPlan.getOperations().stream()
           .map(CompactionOperation::convertFromAvroRecordInstance).collect(toList());
-      LOG.info("CompactionPlanFunction compacting " + operations + " files");
+      LOG.info("CompactionPlanOperator compacting " + operations + " files");
       for (CompactionOperation operation : operations) {
         output.collect(new StreamRecord<>(new CompactionPlanEvent(compactionInstantTime, operation)));
       }