[AMORO-4216] Refactor dangling-delete-files-cleaning via ProcessFactory plugin

amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java

@@ -174,24 +174,6 @@ public class AmoroManagementConf {
           .defaultValue(Duration.ofHours(1))
           .withDescription("Interval for expiring snapshots.");
 
-  public static final ConfigOption<Boolean> CLEAN_DANGLING_DELETE_FILES_ENABLED =
-      ConfigOptions.key("clean-dangling-delete-files.enabled")
-          .booleanType()
-          .defaultValue(true)
-          .withDescription("Enable dangling delete files cleaning.");
-
-  public static final ConfigOption<Integer> CLEAN_DANGLING_DELETE_FILES_THREAD_COUNT =
-      ConfigOptions.key("clean-dangling-delete-files.thread-count")
-          .intType()
-          .defaultValue(10)
-          .withDescription("The number of threads used for dangling delete files cleaning.");
-
-  public static final ConfigOption<Duration> CLEAN_DANGLING_DELETE_FILES_INTERVAL =
-      ConfigOptions.key("clean-dangling-delete-files.interval")
-          .durationType()
-          .defaultValue(Duration.ofDays(1))
-          .withDescription("Interval for cleaning dangling delete files.");
-
   public static final ConfigOption<Boolean> SYNC_HIVE_TABLES_ENABLED =
       ConfigOptions.key("sync-hive-tables.enabled")
           .booleanType()

amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java

@@ -290,7 +290,6 @@ public void startOptimizingService() throws Exception {
     addHandlerChain(optimizingService.getTableRuntimeHandler());
     addHandlerChain(processService.getTableHandlerChain());
     addHandlerChain(InlineTableExecutors.getInstance().getDataExpiringExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getDanglingDeleteFilesCleaningExecutor());
     addHandlerChain(InlineTableExecutors.getInstance().getOptimizingCommitExecutor());
     addHandlerChain(InlineTableExecutors.getInstance().getProcessDataExpiringExecutor());
     addHandlerChain(InlineTableExecutors.getInstance().getBlockerExpiringExecutor());

amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/DanglingDeleteFilesCleaningProcess.java

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.amoro.server.process.iceberg;
+
+import org.apache.amoro.Action;
+import org.apache.amoro.AmoroTable;
+import org.apache.amoro.IcebergActions;
+import org.apache.amoro.TableRuntime;
+import org.apache.amoro.maintainer.TableMaintainer;
+import org.apache.amoro.process.ExecuteEngine;
+import org.apache.amoro.process.LocalProcess;
+import org.apache.amoro.process.TableProcess;
+import org.apache.amoro.server.optimizing.maintainer.TableMaintainerFactory;
+import org.apache.amoro.server.table.DefaultTableRuntime;
+import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+/** Local table process for expiring Iceberg dangling delete files. */
+public class DanglingDeleteFilesCleaningProcess extends TableProcess implements LocalProcess {
+
+  private static final Logger LOG =
+      LoggerFactory.getLogger(DanglingDeleteFilesCleaningProcess.class);
+
+  public DanglingDeleteFilesCleaningProcess(TableRuntime tableRuntime, ExecuteEngine engine) {
+    super(tableRuntime, engine);
+  }
+
+  @Override
+  public String tag() {
+    return getAction().getName().toLowerCase();
+  }
+
+  @Override
+  public void run() {
+    try {
+      AmoroTable<?> amoroTable = tableRuntime.loadTable();
+      TableMaintainer tableMaintainer = TableMaintainerFactory.create(amoroTable, tableRuntime);
+      tableMaintainer.cleanDanglingDeleteFiles();
+      tableRuntime.updateState(
+          DefaultTableRuntime.CLEANUP_STATE_KEY,
+          cleanUp -> cleanUp.setLastDanglingDeleteFilesCleanTime(System.currentTimeMillis()));
+    } catch (Throwable t) {
+      LOG.error(
+          "unexpected dangling delete files cleaning error of table {} ",
+          tableRuntime.getTableIdentifier(),
+          t);
+    }
+  }
+
+  @Override
+  public Action getAction() {
+    return IcebergActions.CLEAN_DANGLING_DELETE;
+  }
+
+  @Override
+  public Map<String, String> getProcessParameters() {
+    return Maps.newHashMap();
+  }
+
+  @Override
+  public Map<String, String> getSummary() {
+    return Maps.newHashMap();
+  }
+}

amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/IcebergProcessFactory.java

@@ -65,6 +65,14 @@ public class IcebergProcessFactory implements ProcessFactory {
           .durationType()
           .defaultValue(Duration.ofDays(1));
 
+  public static final ConfigOption<Boolean> DANGLING_DELETE_FILES_CLEANING_ENABLED =
+      ConfigOptions.key("clean-dangling-delete-files.enabled").booleanType().defaultValue(true);
+
+  public static final ConfigOption<Duration> DANGLING_DELETE_FILES_CLEANING_INTERVAL =
+      ConfigOptions.key("clean-dangling-delete-files.interval")
+          .durationType()
+          .defaultValue(Duration.ofDays(1));
+
   private ExecuteEngine localEngine;
   private final Map<Action, ProcessTriggerStrategy> actions = Maps.newHashMap();
   private final List<TableFormat> formats =
@@ -101,6 +109,8 @@ public Optional<TableProcess> trigger(TableRuntime tableRuntime, Action action)
       return triggerExpireSnapshot(tableRuntime);
     } else if (IcebergActions.DELETE_ORPHANS.equals(action)) {
       return triggerCleanOrphans(tableRuntime);
+    } else if (IcebergActions.CLEAN_DANGLING_DELETE.equals(action)) {
+      return triggerCleanDanglingDelete(tableRuntime);
     }
 
     return Optional.empty();
@@ -130,6 +140,12 @@ public void open(Map<String, String> properties) {
       this.actions.put(
           IcebergActions.DELETE_ORPHANS, ProcessTriggerStrategy.triggerAtFixRate(interval));
     }
+
+    if (configs.getBoolean(DANGLING_DELETE_FILES_CLEANING_ENABLED)) {
+      Duration interval = configs.getDuration(DANGLING_DELETE_FILES_CLEANING_INTERVAL);
+      this.actions.put(
+          IcebergActions.CLEAN_DANGLING_DELETE, ProcessTriggerStrategy.triggerAtFixRate(interval));
+    }
   }
 
   private Optional<TableProcess> triggerExpireSnapshot(TableRuntime tableRuntime) {
@@ -162,6 +178,24 @@ private Optional<TableProcess> triggerCleanOrphans(TableRuntime tableRuntime) {
     return Optional.of(new OrphanFilesCleaningProcess(tableRuntime, localEngine));
   }
 
+  private Optional<TableProcess> triggerCleanDanglingDelete(TableRuntime tableRuntime) {
+    if (localEngine == null
+        || !tableRuntime.getTableConfiguration().isDeleteDanglingDeleteFilesEnabled()) {
+      return Optional.empty();
+    }
+
+    long lastExecuteTime =
+        tableRuntime
+            .getState(DefaultTableRuntime.CLEANUP_STATE_KEY)
+            .getLastDanglingDeleteFilesCleanTime();
+    ProcessTriggerStrategy strategy = actions.get(IcebergActions.CLEAN_DANGLING_DELETE);
+    if (System.currentTimeMillis() - lastExecuteTime < strategy.getTriggerInterval().toMillis()) {
+      return Optional.empty();
+    }
+
+    return Optional.of(new DanglingDeleteFilesCleaningProcess(tableRuntime, localEngine));
+  }
+
   @Override
   public void close() {}
 

amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/InlineTableExecutors.java

@@ -28,7 +28,6 @@ public class InlineTableExecutors {
 
   private static final InlineTableExecutors instance = new InlineTableExecutors();
   private TableRuntimeRefreshExecutor tableRefreshingExecutor;
-  private DanglingDeleteFilesCleaningExecutor danglingDeleteFilesCleaningExecutor;
   private BlockerExpiringExecutor blockerExpiringExecutor;
   private OptimizingCommitExecutor optimizingCommitExecutor;
   private ProcessDataExpiringExecutor processDataExpiringExecutor;
@@ -41,13 +40,6 @@ public static InlineTableExecutors getInstance() {
   }
 
   public void setup(TableService tableService, Configurations conf) {
-    if (conf.getBoolean(AmoroManagementConf.CLEAN_DANGLING_DELETE_FILES_ENABLED)) {
-      this.danglingDeleteFilesCleaningExecutor =
-          new DanglingDeleteFilesCleaningExecutor(
-              tableService,
-              conf.getInteger(AmoroManagementConf.CLEAN_DANGLING_DELETE_FILES_THREAD_COUNT),
-              conf.get(AmoroManagementConf.CLEAN_DANGLING_DELETE_FILES_INTERVAL));
-    }
     this.optimizingCommitExecutor =
         new OptimizingCommitExecutor(
             tableService, conf.getInteger(AmoroManagementConf.OPTIMIZING_COMMIT_THREAD_COUNT));
@@ -100,10 +92,6 @@ public TableRuntimeRefreshExecutor getTableRefreshingExecutor() {
     return tableRefreshingExecutor;
   }
 
-  public DanglingDeleteFilesCleaningExecutor getDanglingDeleteFilesCleaningExecutor() {
-    return danglingDeleteFilesCleaningExecutor;
-  }
-
   public BlockerExpiringExecutor getBlockerExpiringExecutor() {
     return blockerExpiringExecutor;
   }

amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java

@@ -342,8 +342,6 @@ public void beginProcess(OptimizingProcess optimizingProcess) {
   public long getLastCleanTime(CleanupOperation operation) {
     TableRuntimeCleanupState state = store().getState(CLEANUP_STATE_KEY);
     switch (operation) {
-      case DANGLING_DELETE_FILES_CLEANING:
-        return state.getLastDanglingDeleteFilesCleanTime();
       case DATA_EXPIRING:
         return state.getLastDataExpiringTime();
       case SNAPSHOTS_EXPIRING:
@@ -360,9 +358,6 @@ public void updateLastCleanTime(CleanupOperation operation, long time) {
             CLEANUP_STATE_KEY,
             state -> {
               switch (operation) {
-                case DANGLING_DELETE_FILES_CLEANING:
-                  state.setLastDanglingDeleteFilesCleanTime(time);
-                  break;
                 case DATA_EXPIRING:
                   state.setLastDataExpiringTime(time);
                   break;

amoro-ams/src/main/java/org/apache/amoro/server/table/cleanup/CleanupOperation.java

@@ -20,7 +20,6 @@
 
 /** Table cleanup operation enum. Defines different operation types for table cleanup tasks. */
 public enum CleanupOperation {
-  DANGLING_DELETE_FILES_CLEANING,
   DATA_EXPIRING,
   SNAPSHOTS_EXPIRING,
   //  NONE indicates operation types where no cleanup process records are

amoro-ams/src/main/java/org/apache/amoro/server/table/cleanup/TableRuntimeCleanupState.java

@@ -37,8 +37,10 @@ public long getLastDanglingDeleteFilesCleanTime() {
     return lastDanglingDeleteFilesCleanTime;
   }
 
-  public void setLastDanglingDeleteFilesCleanTime(long lastDanglingDeleteFilesCleanTime) {
+  public TableRuntimeCleanupState setLastDanglingDeleteFilesCleanTime(
+      long lastDanglingDeleteFilesCleanTime) {
     this.lastDanglingDeleteFilesCleanTime = lastDanglingDeleteFilesCleanTime;
+    return this;
   }
 
   public long getLastDataExpiringTime() {