[HUDI-4273] Support inline schedule clustering for Flink stream

Author: yuzhaojing

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java

@@ -553,6 +553,24 @@ public void rollbackInflightCompaction(HoodieInstant inflightInstant, Function<S
     getActiveTimeline().revertCompactionInflightToRequested(inflightInstant);
   }
 
+  public void rollbackInflightClustering(HoodieInstant inflightInstant) {
+    rollbackInflightClustering(inflightInstant, s -> Option.empty());
+  }
+
+  /**
+   * Rollback failed clustering. Inflight rollbacks for clustering revert the .inflight file
+   * to the .requested file.
+   *
+   * @param inflightInstant Inflight Clustering Instant
+   */
+  public void rollbackInflightClustering(HoodieInstant inflightInstant, Function<String, Option<HoodiePendingRollbackInfo>> getPendingRollbackInstantFunc) {
+    final String commitTime = getPendingRollbackInstantFunc.apply(inflightInstant.getTimestamp()).map(entry
+        -> entry.getRollbackInstant().getTimestamp()).orElse(HoodieActiveTimeline.createNewInstantTime());
+    scheduleRollback(context, commitTime, inflightInstant, false, config.shouldRollbackUsingMarkers());
+    rollback(context, commitTime, inflightInstant, false, false);
+    getActiveTimeline().revertCompactionInflightToRequested(inflightInstant);
+  }
+
   /**
    * Finalize the written data onto storage. Perform any final cleanups.
    *

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/ScheduleCompactionActionExecutor.java

@@ -71,7 +71,7 @@ public Option<HoodieCompactionPlan> execute() {
     if (!config.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl()
         && !config.getFailedWritesCleanPolicy().isLazy()) {
       // TODO(yihua): this validation is removed for Java client used by kafka-connect.  Need to revisit this.
-      if (config.getEngineType() != EngineType.JAVA) {
+      if (config.getEngineType() == EngineType.SPARK) {
         // if there are inflight writes, their instantTime must not be less than that of compaction instant time
         table.getActiveTimeline().getCommitsTimeline().filterPendingExcludingCompaction().firstInstant()
             .ifPresent(earliestInflight -> ValidationUtils.checkArgument(

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

@@ -594,6 +594,12 @@ private FlinkOptions() {
       .defaultValue(false) // default false for pipeline
       .withDescription("Schedule the cluster plan, default false");
 
+  public static final ConfigOption<Boolean> CLUSTERING_ASYNC_ENABLED = ConfigOptions
+      .key("clustering.async.enabled")
+      .booleanType()
+      .defaultValue(false) // default false for pipeline
+      .withDescription("Async Clustering, default false");
+
   public static final ConfigOption<Integer> CLUSTERING_DELTA_COMMITS = ConfigOptions
       .key("clustering.delta_commits")
       .intType()
@@ -641,7 +647,7 @@ private FlinkOptions() {
   public static final ConfigOption<String> CLUSTERING_SORT_COLUMNS = ConfigOptions
       .key("clustering.plan.strategy.sort.columns")
       .stringType()
-      .noDefaultValue()
+      .defaultValue("")
       .withDescription("Columns to sort the data by when clustering");
 
   public static final ConfigOption<Integer> CLUSTERING_MAX_NUM_GROUPS = ConfigOptions

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java

@@ -20,6 +20,7 @@
 
 import org.apache.hudi.common.model.DefaultHoodieRecordPayload;
 import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.util.StringUtils;
 import org.apache.hudi.index.HoodieIndex;
 import org.apache.hudi.table.format.FilePathUtils;
 
@@ -42,7 +43,10 @@ public static boolean insertClustering(Configuration conf) {
    * Returns whether the insert is clustering disabled with given configuration {@code conf}.
    */
   public static boolean isAppendMode(Configuration conf) {
-    return isCowTable(conf) && isInsertOperation(conf) && !conf.getBoolean(FlinkOptions.INSERT_CLUSTER);
+    // 1. inline clustering is supported for COW table;
+    // 2. async clustering is supported for both COW and MOR table
+    return isCowTable(conf) && isInsertOperation(conf) && !conf.getBoolean(FlinkOptions.INSERT_CLUSTER)
+        || needsScheduleClustering(conf);
   }
 
   /**
@@ -115,4 +119,49 @@ public static boolean emitChangelog(Configuration conf) {
     return conf.getBoolean(FlinkOptions.READ_AS_STREAMING)
         && conf.getBoolean(FlinkOptions.CHANGELOG_ENABLED);
   }
+
+  /**
+   * Returns whether there is need to schedule the async compaction.
+   *
+   * @param conf The flink configuration.
+   */
+  public static boolean needsAsyncCompaction(Configuration conf) {
+    return OptionsResolver.isMorTable(conf)
+        && conf.getBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED);
+  }
+
+  /**
+   * Returns whether there is need to schedule the compaction plan.
+   *
+   * @param conf The flink configuration.
+   */
+  public static boolean needsScheduleCompaction(Configuration conf) {
+    return OptionsResolver.isMorTable(conf)
+        && conf.getBoolean(FlinkOptions.COMPACTION_SCHEDULE_ENABLED);
+  }
+
+  /**
+   * Returns whether there is need to schedule the async clustering.
+   *
+   * @param conf The flink configuration.
+   */
+  public static boolean needsAsyncClustering(Configuration conf) {
+    return isInsertOperation(conf) && conf.getBoolean(FlinkOptions.CLUSTERING_ASYNC_ENABLED);
+  }
+
+  /**
+   * Returns whether there is need to schedule the clustering plan.
+   *
+   * @param conf The flink configuration.
+   */
+  public static boolean needsScheduleClustering(Configuration conf) {
+    return isInsertOperation(conf) && conf.getBoolean(FlinkOptions.CLUSTERING_SCHEDULE_ENABLED);
+  }
+
+  /**
+   * Returns whether the clustering sort is enabled.
+   */
+  public static boolean sortClusteringEnabled(Configuration conf) {
+    return !StringUtils.isNullOrEmpty(conf.getString(FlinkOptions.CLUSTERING_SORT_COLUMNS));
+  }
 }

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

@@ -37,6 +37,7 @@
 import org.apache.hudi.sink.meta.CkpMetadata;
 import org.apache.hudi.sink.utils.HiveSyncContext;
 import org.apache.hudi.sink.utils.NonThrownExecutor;
+import org.apache.hudi.util.ClusteringUtil;
 import org.apache.hudi.util.CompactionUtil;
 import org.apache.hudi.util.StreamerUtil;
 
@@ -253,6 +254,11 @@ public void notifyCheckpointComplete(long checkpointId) {
             CompactionUtil.scheduleCompaction(metaClient, writeClient, tableState.isDeltaTimeCompaction, committed);
           }
 
+          if (tableState.scheduleClustering) {
+            // if async clustering is on, schedule the clustering
+            ClusteringUtil.scheduleClustering(conf, writeClient, committed);
+          }
+
           if (committed) {
             // start new instant.
             startInstant();
@@ -606,6 +612,7 @@ private static class TableState implements Serializable {
     final String commitAction;
     final boolean isOverwrite;
     final boolean scheduleCompaction;
+    final boolean scheduleClustering;
     final boolean syncHive;
     final boolean syncMetadata;
     final boolean isDeltaTimeCompaction;
@@ -615,7 +622,8 @@ private TableState(Configuration conf) {
       this.commitAction = CommitUtils.getCommitActionType(this.operationType,
           HoodieTableType.valueOf(conf.getString(FlinkOptions.TABLE_TYPE).toUpperCase(Locale.ROOT)));
       this.isOverwrite = WriteOperationType.isOverwrite(this.operationType);
-      this.scheduleCompaction = StreamerUtil.needsScheduleCompaction(conf);
+      this.scheduleCompaction = OptionsResolver.needsScheduleCompaction(conf);
+      this.scheduleClustering = OptionsResolver.needsScheduleClustering(conf);
       this.syncHive = conf.getBoolean(FlinkOptions.HIVE_SYNC_ENABLED);
       this.syncMetadata = conf.getBoolean(FlinkOptions.METADATA_ENABLED);
       this.isDeltaTimeCompaction = OptionsResolver.isDeltaTimeCompaction(conf);

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringCommitEvent.java

@@ -24,7 +24,7 @@
 import java.util.List;
 
 /**
- * Represents a commit event from the clustering task {@link ClusteringFunction}.
+ * Represents a commit event from the clustering task {@link ClusteringOperator}.
  */
 public class ClusteringCommitEvent implements Serializable {
   private static final long serialVersionUID = 1L;
@@ -51,6 +51,10 @@ public ClusteringCommitEvent(String instant, List<WriteStatus> writeStatuses, in
     this.taskID = taskID;
   }
 
+  public ClusteringCommitEvent(String instant, int taskID) {
+    this(instant, null, taskID);
+  }
+
   public void setInstant(String instant) {
     this.instant = instant;
   }
@@ -74,4 +78,8 @@ public List<WriteStatus> getWriteStatuses() {
   public int getTaskID() {
     return taskID;
   }
+
+  public boolean isFailed() {
+    return this.writeStatuses == null;
+  }
 }

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringCommitSink.java

@@ -35,6 +35,7 @@
 import org.apache.hudi.sink.CleanFunction;
 import org.apache.hudi.table.HoodieFlinkTable;
 import org.apache.hudi.table.action.HoodieWriteMetadata;
+import org.apache.hudi.util.CompactionUtil;
 import org.apache.hudi.util.StreamerUtil;
 
 import org.apache.flink.configuration.Configuration;
@@ -115,6 +116,30 @@ private void commitIfNecessary(String instant, List<ClusteringCommitEvent> event
     if (!isReady) {
       return;
     }
+
+    if (events.stream().anyMatch(ClusteringCommitEvent::isFailed)) {
+      try {
+        // handle failure case
+        CompactionUtil.rollbackCompaction(table, instant);
+      } finally {
+        // remove commitBuffer to avoid obsolete metadata commit
+        reset(instant);
+      }
+      return;
+    }
+
+    try {
+      doCommit(instant, clusteringPlan, events);
+    } catch (Throwable throwable) {
+      // make it fail-safe
+      LOG.error("Error while committing clustering instant: " + instant, throwable);
+    } finally {
+      // reset the status
+      reset(instant);
+    }
+  }
+
+  private void doCommit(String instant, HoodieClusteringPlan clusteringPlan, List<ClusteringCommitEvent> events) {
     List<WriteStatus> statuses = events.stream()
         .map(ClusteringCommitEvent::getWriteStatuses)
         .flatMap(Collection::stream)
@@ -139,9 +164,6 @@ private void commitIfNecessary(String instant, List<ClusteringCommitEvent> event
     this.table.getMetaClient().reloadActiveTimeline();
     this.writeClient.completeTableService(
         TableServiceType.CLUSTER, writeMetadata.getCommitMetadata().get(), table, instant);
-
-    // reset the status
-    reset(instant);
   }
 
   private void reset(String instant) {