Introduce scheduler module and external resource container

Author: majin.nathan

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

@@ -44,7 +44,7 @@
 import org.apache.amoro.server.resource.ContainerMetadata;
 import org.apache.amoro.server.resource.DefaultOptimizerManager;
 import org.apache.amoro.server.resource.OptimizerManager;
-import org.apache.amoro.server.resource.ResourceContainers;
+import org.apache.amoro.server.resource.InternalContainers;
 import org.apache.amoro.server.table.DefaultTableManager;
 import org.apache.amoro.server.table.DefaultTableService;
 import org.apache.amoro.server.table.RuntimeHandlerChain;
@@ -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/dashboard/controller/OptimizerController.java

@@ -27,7 +27,7 @@
 import org.apache.amoro.server.resource.ContainerMetadata;
 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.resource.InternalContainers;
 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()).releaseOptimizer(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

@@ -32,7 +32,7 @@
 import org.apache.amoro.server.resource.ContainerMetadata;
 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.resource.InternalContainers;
 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()).releaseOptimizer(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

@@ -25,7 +25,7 @@
 import org.apache.amoro.server.dashboard.response.OkResponse;
 import org.apache.amoro.server.resource.ContainerMetadata;
 import org.apache.amoro.server.resource.OptimizerManager;
-import org.apache.amoro.server.resource.ResourceContainers;
+import org.apache.amoro.server.resource.InternalContainers;
 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 renamed to 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";

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);
 

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);
 

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";

amoro-ams/src/main/java/org/apache/amoro/server/resource/ResourceContainers.java renamed to amoro-ams/src/main/java/org/apache/amoro/server/resource/InternalContainers.java

@@ -18,6 +18,7 @@
 
 package org.apache.amoro.server.resource;
 
+import org.apache.amoro.resource.InternalResourceContainer;
 import org.apache.amoro.resource.ResourceContainer;
 import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
@@ -27,15 +28,15 @@
 import java.util.Optional;
 import java.util.stream.Collectors;
 
-public class ResourceContainers {
-  public static final String EXTERNAL_CONTAINER_NAME = "external";
+public class InternalContainers {
+  public static final String UNMANAGED_CONTAINER_NAME = "unmanaged";
   private static final Map<String, ContainerWrapper> globalContainers = Maps.newHashMap();
   private static volatile boolean isInitialized = false;
 
   static {
-    ContainerMetadata metadata = new ContainerMetadata(EXTERNAL_CONTAINER_NAME, "");
+    ContainerMetadata metadata = new ContainerMetadata(UNMANAGED_CONTAINER_NAME, "");
     ContainerWrapper externalContainer = new ContainerWrapper(metadata, null);
-    globalContainers.put(EXTERNAL_CONTAINER_NAME, externalContainer);
+    globalContainers.put(UNMANAGED_CONTAINER_NAME, externalContainer);
   }
 
   public static void init(List<ContainerMetadata> containerList) {
@@ -46,7 +47,7 @@ public static void init(List<ContainerMetadata> containerList) {
     isInitialized = true;
   }
 
-  public static ResourceContainer get(String name) {
+  public static InternalResourceContainer get(String name) {
     checkInitialized();
     return Optional.ofNullable(globalContainers.get(name))
         .map(ContainerWrapper::getContainer)
@@ -70,31 +71,31 @@ public static boolean contains(String name) {
   }
 
   private static class ContainerWrapper {
-    private final ResourceContainer container;
+    private final InternalResourceContainer container;
     private final ContainerMetadata metadata;
 
     public ContainerWrapper(ContainerMetadata metadata) {
       this.metadata = metadata;
       this.container = loadResourceContainer(metadata.getImplClass());
     }
 
-    ContainerWrapper(ContainerMetadata metadata, ResourceContainer container) {
+    ContainerWrapper(ContainerMetadata metadata, InternalResourceContainer container) {
       this.metadata = metadata;
       this.container = container;
     }
 
-    public ResourceContainer getContainer() {
+    public InternalResourceContainer getContainer() {
       return container;
     }
 
     public ContainerMetadata getMetadata() {
       return metadata;
     }
 
-    private ResourceContainer loadResourceContainer(String implClass) {
+    private InternalResourceContainer loadResourceContainer(String implClass) {
       try {
         Class<?> clazz = Class.forName(implClass);
-        ResourceContainer resourceContainer = (ResourceContainer) clazz.newInstance();
+        InternalResourceContainer resourceContainer = (InternalResourceContainer) clazz.newInstance();
         resourceContainer.init(metadata.getName(), metadata.getProperties());
         return resourceContainer;
       } catch (ClassNotFoundException | InstantiationException | IllegalAccessException e) {

amoro-ams/src/test/java/org/apache/amoro/server/AmsEnvironment.java

@@ -32,7 +32,7 @@
 import org.apache.amoro.server.catalog.DefaultCatalogManager;
 import org.apache.amoro.server.catalog.ServerCatalog;
 import org.apache.amoro.server.resource.OptimizerManager;
-import org.apache.amoro.server.resource.ResourceContainers;
+import org.apache.amoro.server.resource.InternalContainers;
 import org.apache.amoro.server.table.DefaultTableService;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
 import org.apache.amoro.shade.guava32.com.google.common.io.MoreFiles;
@@ -310,7 +310,7 @@ public void stopOptimizer() {
         .listOptimizers()
         .forEach(
             resource -> {
-              ResourceContainers.get(resource.getContainerName()).releaseOptimizer(resource);
+              InternalContainers.get(resource.getContainerName()).releaseOptimizer(resource);
             });
   }
 

amoro-common/src/main/java/org/apache/amoro/resource/ExternalResourceContainer.java

@@ -0,0 +1,11 @@
+package org.apache.amoro.resource;
+
+import org.apache.amoro.process.TableProcess;
+import org.apache.amoro.process.TableProcessState;
+
+public interface ExternalResourceContainer {
+
+  Resource submit(TableProcess<TableProcessState> process);
+
+  void release(String resourceId);
+}

amoro-common/src/main/java/org/apache/amoro/resource/InternalResourceContainer.java

@@ -0,0 +1,18 @@
+package org.apache.amoro.resource;
+
+public interface InternalResourceContainer extends ResourceContainer {
+
+  /**
+   * Start a new optimizer.
+   *
+   * @param resource resource information to start the optimizer
+   */
+  void requestResource(Resource resource);
+
+  /**
+   * Release a optimizer
+   *
+   * @param resource resource information to release the optimizer
+   */
+  void releaseOptimizer(Resource resource);
+}

amoro-common/src/main/java/org/apache/amoro/resource/ResourceContainer.java

@@ -31,19 +31,5 @@ public interface ResourceContainer {
    */
   void init(String name, Map<String, String> containerProperties);
 
-  /**
-   * Start a new optimizer.
-   *
-   * @param resource resource information to start the optimizer
-   */
-  void requestResource(Resource resource);
-
-  /**
-   * Release a optimizer
-   *
-   * @param resource resource information to release the optimizer
-   */
-  void releaseOptimizer(Resource resource);
-
   ResourceStatus getStatus(String resourceId);
 }

amoro-scheduler/pom.xml

@@ -0,0 +1,18 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.amoro</groupId>
+        <artifactId>amoro-parent</artifactId>
+        <version>0.8-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>amoro-scheduler</artifactId>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+</project>

pom.xml

@@ -49,6 +49,7 @@
         <module>amoro-common</module>
         <module>amoro-web</module>
         <module>amoro-ams</module>
+        <module>amoro-scheduler</module>
         <module>amoro-optimizer</module>
         <module>amoro-metrics</module>
         <module>amoro-format-iceberg</module>