Merge branch 'master' into lance-format

Author: czy006

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

@@ -53,6 +53,15 @@ public class AmoroManagementConf {
           .defaultValue("admin")
           .withDescription("The administrator password");
 
+  /** Enable master & slave mode, which supports horizontal scaling of AMS. */
+  public static final ConfigOption<Boolean> USE_MASTER_SLAVE_MODE =
+      ConfigOptions.key("use-master-slave-mode")
+          .booleanType()
+          .defaultValue(false)
+          .withDescription(
+              "This setting controls whether to enable the AMS horizontal scaling feature, "
+                  + "which is currently under development and testing.");
+
   public static final ConfigOption<Duration> CATALOG_META_CACHE_EXPIRATION_INTERVAL =
       ConfigOptions.key("catalog-meta-cache.expiration-interval")
           .durationType()

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

@@ -18,6 +18,8 @@
 
 package org.apache.amoro.server;
 
+import static org.apache.amoro.server.AmoroManagementConf.USE_MASTER_SLAVE_MODE;
+
 import io.javalin.Javalin;
 import io.javalin.http.HttpCode;
 import io.javalin.http.staticfiles.Location;
@@ -99,6 +101,7 @@ public class AmoroServiceContainer {
   public static final Logger LOG = LoggerFactory.getLogger(AmoroServiceContainer.class);
 
   public static final String SERVER_CONFIG_FILENAME = "config.yaml";
+  private static boolean IS_MASTER_SLAVE_MODE = false;
 
   private final HighAvailabilityContainer haContainer;
   private DataSource dataSource;
@@ -133,17 +136,24 @@ public static void main(String[] args) {
                     LOG.info("AMS service has been shut down");
                   }));
       service.startRestServices();
-      while (true) {
-        try {
-          // Used to block AMS instances that have not acquired leadership
-          service.waitLeaderShip();
-          service.transitionToLeader();
-          // Used to block AMS instances that have acquired leadership
-          service.waitFollowerShip();
-        } catch (Exception e) {
-          LOG.error("AMS start error", e);
-        } finally {
-          service.transitionToFollower();
+      if (IS_MASTER_SLAVE_MODE) {
+        // Even if one does not become the master, it cannot block the subsequent logic.
+        service.registAndElect();
+        // Regardless of whether tp becomes the master, the service needs to be activated.
+        service.startOptimizingService();
+      } else {
+        while (true) {
+          try {
+            // Used to block AMS instances that have not acquired leadership
+            service.waitLeaderShip();
+            service.transitionToLeader();
+            // Used to block AMS instances that have acquired leadership
+            service.waitFollowerShip();
+          } catch (Exception e) {
+            LOG.error("AMS start error", e);
+          } finally {
+            service.transitionToFollower();
+          }
         }
       }
     } catch (Throwable t) {
@@ -152,6 +162,10 @@ public static void main(String[] args) {
     }
   }
 
+  public void registAndElect() throws Exception {
+    haContainer.registAndElect();
+  }
+
   public enum HAState {
     INITIALIZING(0),
     FOLLOWER(1),
@@ -306,6 +320,7 @@ public void dispose() {
   private void initConfig() throws Exception {
     LOG.info("initializing configurations...");
     new ConfigurationHelper().init();
+    IS_MASTER_SLAVE_MODE = serviceConfig.getBoolean(USE_MASTER_SLAVE_MODE);
   }
 
   public Configurations getServiceConfig() {

amoro-ams/src/main/java/org/apache/amoro/server/ha/DataBaseHighAvailabilityContainer.java

@@ -147,6 +147,9 @@ public void close() {
     }
   }
 
+  @Override
+  public void registAndElect() throws Exception {}
+
   private class HeartbeatRunnable implements Runnable {
     @Override
     public void run() {

amoro-ams/src/main/java/org/apache/amoro/server/ha/HighAvailabilityContainer.java

@@ -42,4 +42,12 @@ public interface HighAvailabilityContainer {
 
   /** Closes the container and releases resources. */
   void close();
+
+  /**
+   * In master-slave mode, this is used for AMS nodes to register and participate in the master
+   * election process.
+   *
+   * @throws Exception If registration fails or participation in the primary election fails.
+   */
+  void registAndElect() throws Exception;
 }

amoro-ams/src/main/java/org/apache/amoro/server/ha/NoopHighAvailabilityContainer.java

@@ -46,4 +46,7 @@ public void waitFollowerShip() throws InterruptedException {
   public void close() {
     LOG.info("Noop HA: closed");
   }
+
+  @Override
+  public void registAndElect() throws Exception {}
 }

amoro-ams/src/main/java/org/apache/amoro/server/ha/ZkHighAvailabilityContainer.java

@@ -140,6 +140,11 @@ public void waitLeaderShip() throws Exception {
     LOG.info("Became the leader of AMS");
   }
 
+  @Override
+  public void registAndElect() throws Exception {
+    // TODO Here you can register for AMS and participate in the election.
+  }
+
   @Override
   public void waitFollowerShip() throws Exception {
     LOG.info("Waiting to become the follower of AMS");

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

@@ -189,6 +189,14 @@ public void before() {
     @AfterEach
     public void after() {
       LOG.info("Test finished.");
+      try {
+        // explicitly clean up possible residual table runtime records
+        if (catalog.tableExists(tableIdentifier)) {
+          catalog.dropTable(tableIdentifier, true);
+        }
+      } catch (Exception e) {
+        LOG.warn("Failed to drop table during cleanup", e);
+      }
       catalog.dropDatabase(database);
     }
 

docs/configuration/ams-config.md

@@ -125,6 +125,7 @@ table td:last-child, table th:last-child { width: 40%; word-break: break-all; }
 | thrift-server.selector-thread-count | 2 | The number of selector threads for the Thrift server. |
 | thrift-server.table-service.bind-port | 1260 | Port that the table service thrift server is bound to. |
 | thrift-server.table-service.worker-thread-count | 20 | The number of worker threads for the Thrift server. |
+| use-master-slave-mode | false | This setting controls whether to enable the AMS horizontal scaling feature, which is currently under development and testing. |
 
 
 ## Shade Utils Configuration