[Subtask]: In master-slave mode, each AMS should automatically senses the optimizer. #3929

Author: wardli

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

@@ -116,34 +116,6 @@ public class AmoroManagementConf {
               "This setting controls whether to enable the AMS horizontal scaling feature, "
                   + "which is currently under development and testing.");
 
-  public static final ConfigOption<Integer> BUCKET_ID_TOTAL_COUNT =
-      ConfigOptions.key("bucket-id.total-count")
-          .intType()
-          .defaultValue(100)
-          .withDescription(
-              "Total count of bucket IDs for assignment. Bucket IDs range from 1 to this value.");
-
-  public static final ConfigOption<Duration> NODE_OFFLINE_TIMEOUT =
-      ConfigOptions.key("node-offline.timeout")
-          .durationType()
-          .defaultValue(Duration.ofMinutes(5))
-          .withDescription(
-              "Timeout duration to determine if a node is offline. After this duration, the node's bucket IDs will be reassigned.");
-
-  public static final ConfigOption<Duration> ASSIGN_INTERVAL =
-      ConfigOptions.key("bucket-assign.interval")
-          .durationType()
-          .defaultValue(Duration.ofSeconds(60))
-          .withDescription(
-              "Interval for bucket assignment service to detect node changes and redistribute bucket IDs.");
-
-  public static final ConfigOption<Duration> BUCKET_TABLE_SYNC_INTERVAL =
-      ConfigOptions.key("bucket-table-sync.interval")
-          .durationType()
-          .defaultValue(Duration.ofSeconds(60))
-          .withDescription(
-              "Interval for syncing tables assigned to bucket IDs in master-slave mode. Each node periodically loads tables from database based on its assigned bucket IDs.");
-
   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

@@ -81,7 +81,6 @@
 import org.apache.amoro.shade.thrift.org.apache.thrift.transport.TTransportException;
 import org.apache.amoro.shade.thrift.org.apache.thrift.transport.TTransportFactory;
 import org.apache.amoro.shade.thrift.org.apache.thrift.transport.layered.TFramedTransport;
-import org.apache.amoro.shade.zookeeper3.org.apache.curator.framework.CuratorFramework;
 import org.apache.amoro.utils.IcebergThreadPools;
 import org.apache.amoro.utils.JacksonUtil;
 import org.apache.commons.lang3.StringUtils;
@@ -273,43 +272,18 @@ public void startOptimizingService() throws Exception {
 
     List<ActionCoordinator> actionCoordinators = defaultRuntimeFactory.supportedCoordinators();
 
-    // In master-slave mode, create BucketAssignStore and AmsAssignService
-    BucketAssignStore bucketAssignStore = null;
-    if (IS_MASTER_SLAVE_MODE && haContainer != null) {
-      String clusterName = serviceConfig.getString(AmoroManagementConf.HA_CLUSTER_NAME);
-      // Choose BucketAssignStore implementation based on HA container type
-      CuratorFramework zkClient = null;
-      if (haContainer instanceof org.apache.amoro.server.ha.ZkHighAvailabilityContainer) {
-        org.apache.amoro.server.ha.ZkHighAvailabilityContainer zkHaContainer =
-            (org.apache.amoro.server.ha.ZkHighAvailabilityContainer) haContainer;
-        zkClient = zkHaContainer.getZkClient();
-        bucketAssignStore = new ZkBucketAssignStore(zkClient, clusterName);
-        LOG.info("Using ZkBucketAssignStore for master-slave mode");
-      } else if (haContainer
-          instanceof org.apache.amoro.server.ha.DataBaseHighAvailabilityContainer) {
-        bucketAssignStore = new DatabaseBucketAssignStore(clusterName);
-        LOG.info("Using DatabaseBucketAssignStore for master-slave mode");
-      } else {
-        LOG.warn(
-            "Unsupported HA container type for master-slave mode: {}",
-            haContainer.getClass().getName());
-      }
-
-      // Create and start AmsAssignService for bucket assignment
-      if (bucketAssignStore != null) {
-        amsAssignService = new AmsAssignService(haContainer, serviceConfig, zkClient);
-        amsAssignService.start();
-        LOG.info("AmsAssignService started for master-slave mode");
-      }
-    }
-
     tableService =
         new DefaultTableService(
             serviceConfig, catalogManager, defaultRuntimeFactory, haContainer, bucketAssignStore);
     processService = new ProcessService(tableService, actionCoordinators, executeEngineManager);
     optimizingService =
         new DefaultOptimizingService(
-            serviceConfig, catalogManager, optimizerManager, tableService, bucketAssignStore, haContainer);
+            serviceConfig,
+            catalogManager,
+            optimizerManager,
+            tableService,
+            bucketAssignStore,
+            haContainer);
 
     LOG.info("Setting up AMS table executors...");
     InlineTableExecutors.getInstance().setup(tableService, serviceConfig);
@@ -693,12 +667,6 @@ private void initContainerConfig() {
           containerProperties.putIfAbsent(
               OptimizerProperties.AMS_OPTIMIZER_URI,
               AmsUtil.getAMSThriftAddress(serviceConfig, Constants.THRIFT_OPTIMIZING_SERVICE_NAME));
-          // Add master-slave mode flag to container properties
-          // Read from serviceConfig directly since IS_MASTER_SLAVE_MODE is set after
-          // initContainerConfig()
-          if (serviceConfig.getBoolean(USE_MASTER_SLAVE_MODE)) {
-            containerProperties.put(OptimizerProperties.OPTIMIZER_MASTER_SLAVE_MODE, "true");
-          }
           // put addition system properties
           container.setProperties(containerProperties);
           containerList.add(container);

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

@@ -39,6 +39,7 @@
 import org.apache.amoro.resource.ResourceType;
 import org.apache.amoro.server.catalog.CatalogManager;
 import org.apache.amoro.server.dashboard.model.OptimizerResourceInfo;
+import org.apache.amoro.server.ha.HighAvailabilityContainer;
 import org.apache.amoro.server.manager.AbstractOptimizerContainer;
 import org.apache.amoro.server.optimizing.OptimizingProcess;
 import org.apache.amoro.server.optimizing.OptimizingQueue;
@@ -56,8 +57,6 @@
 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.BucketAssignStore;
-import org.apache.amoro.server.ha.HighAvailabilityContainer;
 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;
@@ -126,15 +125,6 @@ public class DefaultOptimizingService extends StatedPersistentBase
   private final HighAvailabilityContainer haContainer;
   private final boolean isMasterSlaveMode;
 
-  public DefaultOptimizingService(
-      Configurations serviceConfig,
-      CatalogManager catalogManager,
-      OptimizerManager optimizerManager,
-      TableService tableService,
-      BucketAssignStore bucketAssignStore) {
-    this(serviceConfig, catalogManager, optimizerManager, tableService, bucketAssignStore, null);
-  }
-
   public DefaultOptimizingService(
       Configurations serviceConfig,
       CatalogManager catalogManager,
@@ -579,14 +569,23 @@ public void run() {
       // Use 1/4 of optimizerTouchTimeout as sync interval (default ~30 seconds), used for
       // master-slave follower sync.
       long syncInterval = Math.max(5000, optimizerTouchTimeout / 4);
+      // In non-master-slave mode, this node is always the leader.
+      boolean wasLeader = !isMasterSlaveMode;
       while (!stopped) {
         try {
-          if (isMasterSlaveMode && (haContainer == null || !haContainer.hasLeadership())) {
-            // Not leader: let subclass handle follower state (e.g. sync optimizer list from DB)
-            onFollowerTick(syncInterval);
-          } else {
+          boolean isLeader = !isMasterSlaveMode || haContainer.hasLeadership();
+          if (!wasLeader && isLeader) {
+            // Follower → Leader transition: subclass takes over monitoring of inherited optimizers.
+            onBecomeLeader();
+          }
+          wasLeader = isLeader;
+
+          if (isLeader) {
             T keepingTask = suspendingQueue.take();
             this.processTask(keepingTask);
+          } else {
+            // Not leader: let subclass handle follower state (e.g. sync optimizer list from DB)
+            onFollowerTick(syncInterval);
           }
         } catch (InterruptedException ignored) {
         } catch (Throwable t) {
@@ -600,6 +599,8 @@ public void run() {
     protected void onFollowerTick(long syncInterval) throws InterruptedException {
       Thread.sleep(syncInterval);
     }
+
+    protected void onBecomeLeader() {}
   }
 
   private class OptimizerKeeper extends AbstractKeeper<OptimizerKeepingTask> {
@@ -639,6 +640,17 @@ protected void onFollowerTick(long syncInterval) throws InterruptedException {
       Thread.sleep(syncInterval);
     }
 
+    @Override
+    protected void onBecomeLeader() {
+      LOG.info(
+          "Became leader, starting heartbeat monitoring for {} inherited optimizers",
+          authOptimizers.size());
+      // All optimizers in authOptimizers were loaded from DB by the follower sync loop.
+      // Their touchTime reflects the latest DB-persisted heartbeat, which is the correct
+      // baseline for the new leader's expiry detection.
+      authOptimizers.values().forEach(this::keepInTouch);
+    }
+
     /**
      * Load optimizer information from database. This is used in master-slave mode for follower
      * nodes to sync optimizer state from database. This method performs incremental updates by