Merge branch 'master' into feature/table-summary-without-optimizing

Author: j1wonpark

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

@@ -32,6 +32,8 @@
 import org.apache.amoro.config.Configurations;
 import org.apache.amoro.config.shade.utils.ConfigShadeUtils;
 import org.apache.amoro.exception.AmoroRuntimeException;
+import org.apache.amoro.process.ActionCoordinator;
+import org.apache.amoro.process.ProcessFactory;
 import org.apache.amoro.server.catalog.CatalogManager;
 import org.apache.amoro.server.catalog.DefaultCatalogManager;
 import org.apache.amoro.server.dashboard.DashboardServer;
@@ -47,16 +49,18 @@
 import org.apache.amoro.server.persistence.HttpSessionHandlerFactory;
 import org.apache.amoro.server.persistence.SqlSessionFactoryProvider;
 import org.apache.amoro.server.process.ProcessService;
+import org.apache.amoro.server.process.ProcessService.ExecuteEngineManager;
+import org.apache.amoro.server.process.TableProcessFactoryManager;
 import org.apache.amoro.server.resource.ContainerMetadata;
 import org.apache.amoro.server.resource.Containers;
 import org.apache.amoro.server.resource.DefaultOptimizerManager;
 import org.apache.amoro.server.resource.OptimizerManager;
 import org.apache.amoro.server.scheduler.inline.InlineTableExecutors;
 import org.apache.amoro.server.table.DefaultTableManager;
+import org.apache.amoro.server.table.DefaultTableRuntimeFactory;
 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.TableRuntimeFactoryManager;
 import org.apache.amoro.server.table.TableService;
 import org.apache.amoro.server.terminal.TerminalManager;
 import org.apache.amoro.server.utils.ThriftServiceProxy;
@@ -167,7 +171,7 @@ public static void main(String[] args) {
   }
 
   public void registAndElect() throws Exception {
-    haContainer.registAndElect();
+    haContainer.registerAndElect();
   }
 
   public enum HAState {
@@ -229,17 +233,22 @@ public void transitionToFollower() {
   }
 
   public void startOptimizingService() throws Exception {
-    TableRuntimeFactoryManager tableRuntimeFactoryManager = new TableRuntimeFactoryManager();
-    tableRuntimeFactoryManager.initialize();
+    // Load process factories and build action coordinators from default table runtime factory.
+    TableProcessFactoryManager tableProcessFactoryManager = new TableProcessFactoryManager();
+    tableProcessFactoryManager.initialize();
+    List<ProcessFactory> processFactories = tableProcessFactoryManager.installedPlugins();
 
-    tableService =
-        new DefaultTableService(serviceConfig, catalogManager, tableRuntimeFactoryManager);
+    DefaultTableRuntimeFactory defaultRuntimeFactory = new DefaultTableRuntimeFactory();
+    defaultRuntimeFactory.initialize(processFactories);
 
+    List<ActionCoordinator> actionCoordinators = defaultRuntimeFactory.supportedCoordinators();
+    ExecuteEngineManager executeEngineManager = new ExecuteEngineManager();
+
+    tableService = new DefaultTableService(serviceConfig, catalogManager, defaultRuntimeFactory);
+    processService = new ProcessService(tableService, actionCoordinators, executeEngineManager);
     optimizingService =
         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());

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

@@ -490,16 +490,17 @@ public void run() {
           OptimizerKeepingTask keepingTask = suspendingQueue.take();
           String token = keepingTask.getToken();
           boolean isExpired = !keepingTask.tryKeeping();
+          if (isExpired) {
+            LOG.info("Optimizer {} has been expired, unregister it", keepingTask.getOptimizer());
+            unregisterOptimizer(token);
+          }
           Optional.ofNullable(keepingTask.getQueue())
               .ifPresent(
                   queue ->
                       queue
                           .collectTasks(buildSuspendingPredication(authOptimizers.keySet()))
                           .forEach(task -> retryTask(task, queue)));
-          if (isExpired) {
-            LOG.info("Optimizer {} has been expired, unregister it", keepingTask.getOptimizer());
-            unregisterOptimizer(token);
-          } else {
+          if (!isExpired) {
             LOG.debug("Optimizer {} is being touched, keep it", keepingTask.getOptimizer());
             keepInTouch(keepingTask.getOptimizer());
           }

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

@@ -28,6 +28,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
@@ -135,6 +137,44 @@ public void waitFollowerShip() throws InterruptedException {
     LOG.info("Became the follower of AMS (Database lease)");
   }
 
+  @Override
+  public void registerAndElect() throws Exception {
+    boolean isMasterSlaveMode = serviceConfig.getBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE);
+    if (!isMasterSlaveMode) {
+      LOG.debug("Master-slave mode is not enabled, skip node registration");
+      return;
+    }
+    // In master-slave mode, register node to database by writing OPTIMIZING_SERVICE info
+    // This is similar to ZK mode registering ephemeral nodes
+    long now = System.currentTimeMillis();
+    String optimizingInfoJson = JacksonUtil.toJSONString(optimizingServiceServerInfo);
+    try {
+      doAsIgnoreError(
+          HaLeaseMapper.class,
+          mapper -> {
+            int updated =
+                mapper.updateServerInfo(
+                    clusterName, OPTIMIZING_SERVICE, nodeId, nodeIp, optimizingInfoJson, now);
+            if (updated == 0) {
+              mapper.insertServerInfoIfAbsent(
+                  clusterName, OPTIMIZING_SERVICE, nodeId, nodeIp, optimizingInfoJson, now);
+            }
+          });
+      LOG.info(
+          "Registered AMS node to database: nodeId={}, optimizingService={}",
+          nodeId,
+          optimizingServiceServerInfo);
+    } catch (Exception e) {
+      LOG.error("Failed to register node to database", e);
+      throw e;
+    }
+  }
+
+  @Override
+  public boolean hasLeadership() {
+    return isLeader.get();
+  }
+
   /** Closes the heartbeat executor safely. */
   @Override
   public void close() {
@@ -147,9 +187,6 @@ public void close() {
     }
   }
 
-  @Override
-  public void registAndElect() throws Exception {}
-
   private class HeartbeatRunnable implements Runnable {
     @Override
     public void run() {
@@ -304,6 +341,40 @@ private void onLeaderLost() {
     }
   }
 
+  @Override
+  public List<AmsServerInfo> getAliveNodes() {
+    List<AmsServerInfo> aliveNodes = new ArrayList<>();
+    if (!isLeader.get()) {
+      LOG.warn("Only leader node can get alive nodes list");
+      return aliveNodes;
+    }
+    try {
+      long currentTime = System.currentTimeMillis();
+      List<HaLeaseMeta> leases =
+          getAs(
+              HaLeaseMapper.class,
+              mapper -> mapper.selectLeasesByService(clusterName, OPTIMIZING_SERVICE));
+      for (HaLeaseMeta lease : leases) {
+        // Only include nodes with valid (non-expired) leases
+        if (lease.getLeaseExpireTs() != null && lease.getLeaseExpireTs() > currentTime) {
+          if (lease.getServerInfoJson() != null && !lease.getServerInfoJson().isEmpty()) {
+            try {
+              AmsServerInfo nodeInfo =
+                  JacksonUtil.parseObject(lease.getServerInfoJson(), AmsServerInfo.class);
+              aliveNodes.add(nodeInfo);
+            } catch (Exception e) {
+              LOG.warn("Failed to parse server info for node {}", lease.getNodeId(), e);
+            }
+          }
+        }
+      }
+    } catch (Exception e) {
+      LOG.error("Failed to get alive nodes from database", e);
+      throw e;
+    }
+    return aliveNodes;
+  }
+
   private AmsServerInfo buildServerInfo(String host, int thriftBindPort, int restBindPort) {
     AmsServerInfo amsServerInfo = new AmsServerInfo();
     amsServerInfo.setHost(host);

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

@@ -18,6 +18,10 @@
 
 package org.apache.amoro.server.ha;
 
+import org.apache.amoro.client.AmsServerInfo;
+
+import java.util.List;
+
 /**
  * Common interface for high availability (HA) containers.
  *
@@ -49,5 +53,19 @@ public interface HighAvailabilityContainer {
    *
    * @throws Exception If registration fails or participation in the primary election fails.
    */
-  void registAndElect() throws Exception;
+  void registerAndElect() throws Exception;
+
+  /**
+   * Used in master-slave mode to obtain information about all currently registered AMS nodes.
+   *
+   * @return List<AmsServerInfo>
+   */
+  List<AmsServerInfo> getAliveNodes();
+
+  /**
+   * Used to determine whether the current AMS node is the primary node.
+   *
+   * @return true if the current AMS node is the primary node, false otherwise
+   */
+  boolean hasLeadership();
 }

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

@@ -18,9 +18,11 @@
 
 package org.apache.amoro.server.ha;
 
+import org.apache.amoro.client.AmsServerInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 
 /** No-op HA container that never blocks and performs no leader election. */
@@ -48,5 +50,15 @@ public void close() {
   }
 
   @Override
-  public void registAndElect() throws Exception {}
+  public void registerAndElect() throws Exception {}
+
+  @Override
+  public List<AmsServerInfo> getAliveNodes() {
+    return List.of();
+  }
+
+  @Override
+  public boolean hasLeadership() {
+    return false;
+  }
 }