Revert master

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/ClusteringPlanActionExecutor.java

@@ -21,10 +21,8 @@
 import org.apache.hudi.avro.model.HoodieClusteringPlan;
 import org.apache.hudi.avro.model.HoodieRequestedReplaceMetadata;
 import org.apache.hudi.common.engine.HoodieEngineContext;
-import org.apache.hudi.common.model.ActionType;
 import org.apache.hudi.common.model.HoodieRecordPayload;
 import org.apache.hudi.common.model.WriteOperationType;
-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.table.timeline.TimelineMetadataUtils;
@@ -41,9 +39,7 @@
 
 import java.io.IOException;
 import java.util.Collections;
-import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 public class ClusteringPlanActionExecutor<T extends HoodieRecordPayload, I, K, O> extends BaseActionExecutor<T, I, K, O, Option<HoodieClusteringPlan>> {
 
@@ -106,22 +102,6 @@ public Option<HoodieClusteringPlan> execute() {
         throw new HoodieIOException("Exception scheduling clustering", ioe);
       }
     }
-
-    if (config.isTableManagerEnabled() && config.getTableManagerConfig().getTableManagerActions().contains(ActionType.replacecommit.name())) {
-      submitClusteringToService();
-    }
-
     return planOption;
   }
-
-  private void submitClusteringToService() {
-    HoodieTableMetaClient metaClient = table.getMetaClient();
-    List<String> instantsToSubmit = metaClient.getActiveTimeline()
-        .filterPendingReplaceTimeline()
-        .getInstants()
-        .map(HoodieInstant::getTimestamp)
-        .collect(Collectors.toList());
-    HoodieTableManagerClient tableManagerClient = new HoodieTableManagerClient(metaClient, config.getTableManagerConfig());
-    tableManagerClient.submitClustering(instantsToSubmit);
-  }
 }

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java

@@ -333,10 +333,6 @@ protected void completeCompaction(HoodieCommitMetadata metadata,
   protected HoodieWriteMetadata<JavaRDD<WriteStatus>> compact(String compactionInstantTime, boolean shouldComplete) {
     HoodieSparkTable<T> table = HoodieSparkTable.create(config, context);
     preWrite(compactionInstantTime, WriteOperationType.COMPACT, table.getMetaClient());
-    // do not compact a complete instant.
-    if (table.getActiveTimeline().filterCompletedInstants().containsInstant(compactionInstantTime)) {
-      return null;
-    }
     HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
     HoodieInstant inflightInstant = HoodieTimeline.getCompactionInflightInstant(compactionInstantTime);
     if (pendingCompactionTimeline.containsInstant(inflightInstant)) {