[Refactor] Introduce Iceberg TableRuntimePlugin

Author: majin1102

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

@@ -54,9 +54,10 @@
 import org.apache.amoro.server.scheduler.inline.InlineTableExecutors;
 import org.apache.amoro.server.table.DefaultTableManager;
 import org.apache.amoro.server.table.DefaultTableService;
-import org.apache.amoro.server.table.RuntimeHandlerChain;
+import org.apache.amoro.server.table.IcebergTablePlugin;
 import org.apache.amoro.server.table.TableManager;
 import org.apache.amoro.server.table.TableRuntimeFactoryManager;
+import org.apache.amoro.server.table.TableRuntimePlugin;
 import org.apache.amoro.server.table.TableService;
 import org.apache.amoro.server.terminal.TerminalManager;
 import org.apache.amoro.server.utils.ThriftServiceProxy;
@@ -198,7 +199,7 @@ public void waitFollowerShip() throws Exception {
     haContainer.waitFollowerShip();
   }
 
-  public void startRestServices() throws Exception {
+  public void startRestServices() {
     EventsManager.getInstance();
     MetricManager.getInstance();
 
@@ -239,33 +240,32 @@ public void startOptimizingService() throws Exception {
         new DefaultOptimizingService(serviceConfig, catalogManager, optimizerManager, tableService);
 
     processService = new ProcessService(serviceConfig, tableService);
-
-    LOG.info("Setting up AMS table executors...");
-    InlineTableExecutors.getInstance().setup(tableService, serviceConfig);
-    addHandlerChain(optimizingService.getTableRuntimeHandler());
-    addHandlerChain(processService.getTableHandlerChain());
-    addHandlerChain(InlineTableExecutors.getInstance().getDataExpiringExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getSnapshotsExpiringExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getOrphanFilesCleaningExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getDanglingDeleteFilesCleaningExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getOptimizingCommitExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getOptimizingExpiringExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getBlockerExpiringExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getHiveCommitSyncExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getTableRefreshingExecutor());
-    addHandlerChain(InlineTableExecutors.getInstance().getTagsAutoCreatingExecutor());
-    tableService.initialize();
+    tableService.initialize(initTablePlugins());
     LOG.info("AMS table service have been initialized");
-    tableManager.setTableService(tableService);
 
     initThriftService();
     startThriftService();
   }
 
-  private void addHandlerChain(RuntimeHandlerChain chain) {
-    if (chain != null) {
-      tableService.addHandlerChain(chain);
-    }
+  private List<TableRuntimePlugin> initTablePlugins() {
+    LOG.info("Setting up AMS table executors...");
+    InlineTableExecutors.getInstance().setup(tableService, serviceConfig);
+    IcebergTablePlugin icebergTablePlugin =
+        IcebergTablePlugin.builder()
+            .addHandler(optimizingService.getTableRuntimeHandler())
+            .addHandler(processService.getTableHandlerChain())
+            .addHandler(InlineTableExecutors.getInstance().getDataExpiringExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getSnapshotsExpiringExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getOrphanFilesCleaningExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getDanglingDeleteFilesCleaningExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getOptimizingCommitExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getOptimizingExpiringExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getBlockerExpiringExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getHiveCommitSyncExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getTableRefreshingExecutor())
+            .addHandler(InlineTableExecutors.getInstance().getTagsAutoCreatingExecutor())
+            .build();
+    return List.of(icebergTablePlugin);
   }
 
   public void disposeOptimizingService() {

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

@@ -114,7 +114,7 @@ public void dropTableMetadata(TableIdentifier tableIdentifier, boolean deleteDat
     }
 
     ServerTableIdentifier serverTableIdentifier = internalCatalog.dropTable(database, table);
-    tableService().ifPresent(s -> s.onTableDropped(internalCatalog, serverTableIdentifier));
+    tableService().ifPresent(s -> s.removeTable(serverTableIdentifier));
   }
 
   @Override
@@ -128,7 +128,7 @@ public void createTable(String catalogName, TableMetadata tableMetadata) {
     }
 
     TableMetadata metadata = catalog.createTable(tableMetadata);
-    tableService().ifPresent(s -> s.onTableCreated(catalog, metadata.getTableIdentifier()));
+    tableService().ifPresent(s -> s.addTable(metadata));
   }
 
   @Override

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

@@ -42,7 +42,7 @@ public String name() {
   }
 
   @Override
-  public Optional<TableRuntimeCreator> accept(
+  public Optional<Creator> accept(
       ServerTableIdentifier tableIdentifier, Map<String, String> tableProperties) {
     if (tableIdentifier
         .getFormat()
@@ -52,7 +52,7 @@ public Optional<TableRuntimeCreator> accept(
     return Optional.empty();
   }
 
-  private static class TableRuntimeCreatorImpl implements TableRuntimeFactory.TableRuntimeCreator {
+  private static class TableRuntimeCreatorImpl implements TableRuntimeFactory.Creator {
     @Override
     public List<StateKey<?>> requiredStateKeys() {
       return DefaultTableRuntime.REQUIRED_STATES;