[AMORO-3485] Introduce scheduler module and external resource container

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

@@ -43,14 +43,14 @@
 import org.apache.amoro.server.persistence.SqlSessionFactoryProvider;
 import org.apache.amoro.server.resource.ContainerMetadata;
 import org.apache.amoro.server.resource.DefaultOptimizerManager;
+import org.apache.amoro.server.resource.InternalContainers;
 import org.apache.amoro.server.resource.OptimizerManager;
-import org.apache.amoro.server.resource.ResourceContainers;
+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.TableManager;
 import org.apache.amoro.server.table.TableService;
-import org.apache.amoro.server.table.executor.AsyncTableExecutors;
 import org.apache.amoro.server.terminal.TerminalManager;
 import org.apache.amoro.server.utils.ThriftServiceProxy;
 import org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting;
@@ -165,18 +165,18 @@ public void startService() throws Exception {
         new DefaultOptimizingService(serviceConfig, catalogManager, optimizerManager, tableService);
 
     LOG.info("Setting up AMS table executors...");
-    AsyncTableExecutors.getInstance().setup(tableService, serviceConfig);
+    InlineTableExecutors.getInstance().setup(tableService, serviceConfig);
     addHandlerChain(optimizingService.getTableRuntimeHandler());
-    addHandlerChain(AsyncTableExecutors.getInstance().getDataExpiringExecutor());
-    addHandlerChain(AsyncTableExecutors.getInstance().getSnapshotsExpiringExecutor());
-    addHandlerChain(AsyncTableExecutors.getInstance().getOrphanFilesCleaningExecutor());
-    addHandlerChain(AsyncTableExecutors.getInstance().getDanglingDeleteFilesCleaningExecutor());
-    addHandlerChain(AsyncTableExecutors.getInstance().getOptimizingCommitExecutor());
-    addHandlerChain(AsyncTableExecutors.getInstance().getOptimizingExpiringExecutor());
-    addHandlerChain(AsyncTableExecutors.getInstance().getBlockerExpiringExecutor());
-    addHandlerChain(AsyncTableExecutors.getInstance().getHiveCommitSyncExecutor());
-    addHandlerChain(AsyncTableExecutors.getInstance().getTableRefreshingExecutor());
-    addHandlerChain(AsyncTableExecutors.getInstance().getTagsAutoCreatingExecutor());
+    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();
     LOG.info("AMS table service have been initialized");
     tableManager.setTableService(tableService);
@@ -528,7 +528,7 @@ private void initContainerConfig() {
           containerList.add(container);
         }
       }
-      ResourceContainers.init(containerList);
+      InternalContainers.init(containerList);
     }
   }
 

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

@@ -47,8 +47,8 @@
 import org.apache.amoro.server.resource.OptimizerManager;
 import org.apache.amoro.server.resource.OptimizerThread;
 import org.apache.amoro.server.resource.QuotaProvider;
+import org.apache.amoro.server.table.DefaultTableRuntime;
 import org.apache.amoro.server.table.RuntimeHandlerChain;
-import org.apache.amoro.server.table.TableRuntime;
 import org.apache.amoro.server.table.TableService;
 import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Sets;
@@ -136,17 +136,19 @@ public RuntimeHandlerChain getTableRuntimeHandler() {
     return tableHandlerChain;
   }
 
-  private void loadOptimizingQueues(List<TableRuntime> tableRuntimeMetaList) {
+  private void loadOptimizingQueues(List<DefaultTableRuntime> tableRuntimeList) {
     List<ResourceGroup> optimizerGroups =
         getAs(ResourceMapper.class, ResourceMapper::selectResourceGroups);
     List<OptimizerInstance> optimizers = getAs(OptimizerMapper.class, OptimizerMapper::selectAll);
-    Map<String, List<TableRuntime>> groupToTableRuntimes =
-        tableRuntimeMetaList.stream()
-            .collect(Collectors.groupingBy(TableRuntime::getOptimizerGroup));
+    Map<String, List<DefaultTableRuntime>> groupToTableRuntimes =
+        tableRuntimeList.stream()
+            .collect(
+                Collectors.groupingBy(
+                    tableRuntime -> tableRuntime.getOptimizingState().getOptimizerGroup()));
     optimizerGroups.forEach(
         group -> {
           String groupName = group.getName();
-          List<TableRuntime> tableRuntimes = groupToTableRuntimes.remove(groupName);
+          List<DefaultTableRuntime> tableRuntimes = groupToTableRuntimes.remove(groupName);
           OptimizingQueue optimizingQueue =
               new OptimizingQueue(
                   catalogManager,
@@ -283,11 +285,11 @@ public boolean cancelProcess(long processId) throws TException {
       return false;
     }
     long tableId = processMeta.getTableId();
-    TableRuntime tableRuntime = tableService.getRuntime(tableId);
+    DefaultTableRuntime tableRuntime = tableService.getRuntime(tableId);
     if (tableRuntime == null) {
       return false;
     }
-    OptimizingProcess process = tableRuntime.getOptimizingProcess();
+    OptimizingProcess process = tableRuntime.getOptimizingState().getOptimizingProcess();
     if (process == null || process.getProcessId() != processId) {
       return false;
     }
@@ -372,37 +374,39 @@ public int getTotalQuota(String resourceGroup) {
   private class TableRuntimeHandlerImpl extends RuntimeHandlerChain {
 
     @Override
-    public void handleStatusChanged(TableRuntime tableRuntime, OptimizingStatus originalStatus) {
-      if (!tableRuntime.getOptimizingStatus().isProcessing()) {
-        getOptionalQueueByGroup(tableRuntime.getOptimizerGroup())
+    public void handleStatusChanged(
+        DefaultTableRuntime tableRuntime, OptimizingStatus originalStatus) {
+      if (!tableRuntime.getOptimizingState().getOptimizingStatus().isProcessing()) {
+        getOptionalQueueByGroup(tableRuntime.getOptimizingState().getOptimizerGroup())
             .ifPresent(q -> q.refreshTable(tableRuntime));
       }
     }
 
     @Override
-    public void handleConfigChanged(TableRuntime tableRuntime, TableConfiguration originalConfig) {
+    public void handleConfigChanged(
+        DefaultTableRuntime tableRuntime, TableConfiguration originalConfig) {
       String originalGroup = originalConfig.getOptimizingConfig().getOptimizerGroup();
-      if (!tableRuntime.getOptimizerGroup().equals(originalGroup)) {
+      if (!tableRuntime.getOptimizingState().getOptimizerGroup().equals(originalGroup)) {
         getOptionalQueueByGroup(originalGroup).ifPresent(q -> q.releaseTable(tableRuntime));
       }
-      getOptionalQueueByGroup(tableRuntime.getOptimizerGroup())
+      getOptionalQueueByGroup(tableRuntime.getOptimizingState().getOptimizerGroup())
           .ifPresent(q -> q.refreshTable(tableRuntime));
     }
 
     @Override
-    public void handleTableAdded(AmoroTable<?> table, TableRuntime tableRuntime) {
-      getOptionalQueueByGroup(tableRuntime.getOptimizerGroup())
+    public void handleTableAdded(AmoroTable<?> table, DefaultTableRuntime tableRuntime) {
+      getOptionalQueueByGroup(tableRuntime.getOptimizingState().getOptimizerGroup())
           .ifPresent(q -> q.refreshTable(tableRuntime));
     }
 
     @Override
-    public void handleTableRemoved(TableRuntime tableRuntime) {
-      getOptionalQueueByGroup(tableRuntime.getOptimizerGroup())
+    public void handleTableRemoved(DefaultTableRuntime tableRuntime) {
+      getOptionalQueueByGroup(tableRuntime.getOptimizingState().getOptimizerGroup())
           .ifPresent(queue -> queue.releaseTable(tableRuntime));
     }
 
     @Override
-    protected void initHandler(List<TableRuntime> tableRuntimeList) {
+    protected void initHandler(List<DefaultTableRuntime> tableRuntimeList) {
       LOG.info("OptimizerManagementService begin initializing");
       loadOptimizingQueues(tableRuntimeList);
       optimizerKeeper.start();

amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/OptimizerController.java

@@ -25,9 +25,9 @@
 import org.apache.amoro.server.DefaultOptimizingService;
 import org.apache.amoro.server.dashboard.response.OkResponse;
 import org.apache.amoro.server.resource.ContainerMetadata;
+import org.apache.amoro.server.resource.InternalContainers;
 import org.apache.amoro.server.resource.OptimizerInstance;
 import org.apache.amoro.server.resource.OptimizerManager;
-import org.apache.amoro.server.resource.ResourceContainers;
 import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
 
 import java.util.List;
@@ -67,7 +67,7 @@ public void releaseOptimizer(Context ctx) {
             "The resource ID %s has not been indexed" + " to any optimizer.", resourceId));
     Resource resource = optimizerManager.getResource(resourceId);
     resource.getProperties().putAll(optimizerInstances.get(0).getProperties());
-    ResourceContainers.get(resource.getContainerName()).releaseOptimizer(resource);
+    InternalContainers.get(resource.getContainerName()).releaseResource(resource);
     optimizerManager.deleteResource(resourceId);
     optimizerManager.deleteOptimizer(resource.getGroupName(), resourceId);
     ctx.json(OkResponse.of("Success to release optimizer"));
@@ -85,7 +85,7 @@ public void createOptimizer(Context ctx) {
             .setProperties(resourceGroup.getProperties())
             .setThreadCount(parallelism)
             .build();
-    ResourceContainers.get(resource.getContainerName()).requestResource(resource);
+    InternalContainers.get(resource.getContainerName()).requestResource(resource);
     optimizerManager.createResource(resource);
     ctx.json(OkResponse.of("success to create optimizer"));
   }
@@ -94,7 +94,7 @@ public void createOptimizer(Context ctx) {
   public void getContainers(Context ctx) {
     ctx.json(
         OkResponse.of(
-            ResourceContainers.getMetadataList().stream()
+            InternalContainers.getMetadataList().stream()
                 .map(ContainerMetadata::getName)
                 .collect(Collectors.toList())));
   }

amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/OptimizerGroupController.java

@@ -30,9 +30,9 @@
 import org.apache.amoro.server.dashboard.response.PageResult;
 import org.apache.amoro.server.optimizing.OptimizingStatus;
 import org.apache.amoro.server.resource.ContainerMetadata;
+import org.apache.amoro.server.resource.InternalContainers;
 import org.apache.amoro.server.resource.OptimizerInstance;
 import org.apache.amoro.server.resource.OptimizerManager;
-import org.apache.amoro.server.resource.ResourceContainers;
 import org.apache.amoro.server.table.TableManager;
 import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
 import org.apache.commons.lang3.StringUtils;
@@ -159,7 +159,7 @@ public void getOptimizerGroups(Context ctx) {
         optimizerManager.listResourceGroups().stream()
             .filter(
                 resourceGroup ->
-                    !ResourceContainers.EXTERNAL_CONTAINER_NAME.equals(
+                    !InternalContainers.UNMANAGED_CONTAINER_NAME.equals(
                         resourceGroup.getContainer()))
             .map(
                 e -> {
@@ -209,7 +209,7 @@ public void releaseOptimizer(Context ctx) {
             "The resource ID %s has not been indexed" + " to any optimizer.", resourceId));
     Resource resource = optimizerManager.getResource(resourceId);
     resource.getProperties().putAll(optimizerInstances.get(0).getProperties());
-    ResourceContainers.get(resource.getContainerName()).releaseOptimizer(resource);
+    InternalContainers.get(resource.getContainerName()).releaseResource(resource);
     optimizerManager.deleteResource(resourceId);
     optimizerManager.deleteOptimizer(resource.getGroupName(), resourceId);
     ctx.json(OkResponse.of("Success to release optimizer"));
@@ -228,7 +228,7 @@ public void scaleOutOptimizer(Context ctx) {
             .setProperties(resourceGroup.getProperties())
             .setThreadCount(parallelism)
             .build();
-    ResourceContainers.get(resource.getContainerName()).requestResource(resource);
+    InternalContainers.get(resource.getContainerName()).requestResource(resource);
     optimizerManager.createResource(resource);
     ctx.json(OkResponse.of("success to scaleOut optimizer"));
   }
@@ -303,7 +303,7 @@ public void deleteCheckResourceGroup(Context ctx) {
   public void getContainers(Context ctx) {
     ctx.json(
         OkResponse.of(
-            ResourceContainers.getMetadataList().stream()
+            InternalContainers.getMetadataList().stream()
                 .map(ContainerMetadata::getName)
                 .collect(Collectors.toList())));
   }

amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/SettingController.java

@@ -24,8 +24,8 @@
 import org.apache.amoro.server.AmoroManagementConf;
 import org.apache.amoro.server.dashboard.response.OkResponse;
 import org.apache.amoro.server.resource.ContainerMetadata;
+import org.apache.amoro.server.resource.InternalContainers;
 import org.apache.amoro.server.resource.OptimizerManager;
-import org.apache.amoro.server.resource.ResourceContainers;
 import org.glassfish.jersey.internal.guava.Sets;
 
 import java.util.ArrayList;
@@ -72,7 +72,7 @@ private void putSetting(Map<String, String> settingMap, String key, Object value
 
   /** Get container settings. */
   public void getContainerSetting(Context ctx) {
-    List<ContainerMetadata> containerMetas = ResourceContainers.getMetadataList();
+    List<ContainerMetadata> containerMetas = InternalContainers.getMetadataList();
     List<Map<String, Object>> result = new ArrayList<>();
     Objects.requireNonNull(containerMetas)
         .forEach(

amoro-ams/src/main/java/org/apache/amoro/server/manager/AbstractResourceContainer.java → amoro-ams/src/main/java/org/apache/amoro/server/manager/AbstractOptimizerContainer.java

@@ -19,8 +19,8 @@
 package org.apache.amoro.server.manager;
 
 import org.apache.amoro.OptimizerProperties;
+import org.apache.amoro.resource.InternalResourceContainer;
 import org.apache.amoro.resource.Resource;
-import org.apache.amoro.resource.ResourceContainer;
 import org.apache.amoro.resource.ResourceStatus;
 import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
 import org.apache.commons.lang3.StringUtils;
@@ -30,7 +30,7 @@
 import java.util.List;
 import java.util.Map;
 
-public abstract class AbstractResourceContainer implements ResourceContainer {
+public abstract class AbstractOptimizerContainer implements InternalResourceContainer {
   private String containerName;
   private Map<String, String> containerProperties;
 

amoro-ams/src/main/java/org/apache/amoro/server/manager/FlinkOptimizerContainer.java

@@ -79,7 +79,7 @@
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
-public class FlinkOptimizerContainer extends AbstractResourceContainer {
+public class FlinkOptimizerContainer extends AbstractOptimizerContainer {
   private static final Logger LOG = LoggerFactory.getLogger(FlinkOptimizerContainer.class);
 
   public static final String FLINK_HOME_PROPERTY = "flink-home";
@@ -385,7 +385,7 @@ private <T> T fetchCommandOutput(Process exec, Function<String, T> commandReader
   }
 
   @Override
-  public void releaseOptimizer(Resource resource) {
+  public void releaseResource(Resource resource) {
     if (target.runByFlinkRestClient()) {
       Preconditions.checkArgument(
           resource.getProperties().containsKey(SESSION_CLUSTER_JOB_ID),

amoro-ams/src/main/java/org/apache/amoro/server/manager/KubernetesOptimizerContainer.java

@@ -56,7 +56,7 @@
 import java.util.stream.Collectors;
 
 /** Kubernetes Optimizer Container with Standalone Optimizer */
-public class KubernetesOptimizerContainer extends AbstractResourceContainer {
+public class KubernetesOptimizerContainer extends AbstractOptimizerContainer {
 
   private static final Logger LOG = LoggerFactory.getLogger(KubernetesOptimizerContainer.class);
 
@@ -348,7 +348,7 @@ public Deployment initPodTemplateFromFrontEnd(
   }
 
   @Override
-  public void releaseOptimizer(Resource resource) {
+  public void releaseResource(Resource resource) {
     String resourceId = resource.getResourceId();
     LOG.info("release Kubernetes Optimizer Container {}", resourceId);
     String namespace = resource.getProperties().get(NAMESPACE);

amoro-ams/src/main/java/org/apache/amoro/server/manager/LocalOptimizerContainer.java

@@ -27,7 +27,7 @@
 import java.util.Collections;
 import java.util.Map;
 
-public class LocalOptimizerContainer extends AbstractResourceContainer {
+public class LocalOptimizerContainer extends AbstractOptimizerContainer {
 
   private static final Logger LOG = LoggerFactory.getLogger(LocalOptimizerContainer.class);
 
@@ -65,7 +65,7 @@ protected String buildOptimizerStartupArgsString(Resource resource) {
   }
 
   @Override
-  public void releaseOptimizer(Resource resource) {
+  public void releaseResource(Resource resource) {
     long jobId = Long.parseLong(resource.getRequiredProperty(Resource.PROPERTY_JOB_ID));
 
     String os = System.getProperty("os.name").toLowerCase();

amoro-ams/src/main/java/org/apache/amoro/server/manager/SparkOptimizerContainer.java

@@ -40,7 +40,7 @@
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
-public class SparkOptimizerContainer extends AbstractResourceContainer {
+public class SparkOptimizerContainer extends AbstractOptimizerContainer {
   private static final Logger LOG = LoggerFactory.getLogger(SparkOptimizerContainer.class);
 
   public static final String SPARK_HOME_PROPERTY = "spark-home";
@@ -247,7 +247,7 @@ private <T> T fetchCommandOutput(Process exec, Function<String, T> commandReader
   }
 
   @Override
-  public void releaseOptimizer(Resource resource) {
+  public void releaseResource(Resource resource) {
     String releaseCommand;
     if (deployedOnKubernetes()) {
       releaseCommand = buildReleaseKubernetesCommand(resource);

amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizerGroupMetrics.java

@@ -145,39 +145,39 @@ public void register() {
         (Gauge<Long>)
             () ->
                 optimizingQueue.getSchedulingPolicy().getTableRuntimeMap().values().stream()
-                    .filter(t -> t.getOptimizingStatus().equals(PLANNING))
+                    .filter(t -> t.getOptimizingState().getOptimizingStatus().equals(PLANNING))
                     .count());
     registerMetric(
         registry,
         OPTIMIZER_GROUP_PENDING_TABLES,
         (Gauge<Long>)
             () ->
                 optimizingQueue.getSchedulingPolicy().getTableRuntimeMap().values().stream()
-                    .filter(t -> t.getOptimizingStatus().equals(PENDING))
+                    .filter(t -> t.getOptimizingState().getOptimizingStatus().equals(PENDING))
                     .count());
     registerMetric(
         registry,
         OPTIMIZER_GROUP_EXECUTING_TABLES,
         (Gauge<Long>)
             () ->
                 optimizingQueue.getSchedulingPolicy().getTableRuntimeMap().values().stream()
-                    .filter(t -> t.getOptimizingStatus().isProcessing())
+                    .filter(t -> t.getOptimizingState().getOptimizingStatus().isProcessing())
                     .count());
     registerMetric(
         registry,
         OPTIMIZER_GROUP_IDLE_TABLES,
         (Gauge<Long>)
             () ->
                 optimizingQueue.getSchedulingPolicy().getTableRuntimeMap().values().stream()
-                    .filter(t -> t.getOptimizingStatus().equals(IDLE))
+                    .filter(t -> t.getOptimizingState().getOptimizingStatus().equals(IDLE))
                     .count());
     registerMetric(
         registry,
         OPTIMIZER_GROUP_COMMITTING_TABLES,
         (Gauge<Long>)
             () ->
                 optimizingQueue.getSchedulingPolicy().getTableRuntimeMap().values().stream()
-                    .filter(t -> t.getOptimizingStatus().equals(COMMITTING))
+                    .filter(t -> t.getOptimizingState().getOptimizingStatus().equals(COMMITTING))
                     .count());
 
     registerMetric(