[fix](cloud) fix active-priority rebalancer starvation (apache#60835)

Root cause:
- In active-priority mode, performBalancing() returns early when active
objects are still unbalanced. Under sustained hot-tablet pressure, this
keeps happening and INACTIVE_ONLY never runs. Inactive partitions/tables
are starved, so BE scale-out rebalancing becomes very slow.
- runAfterCatalogReady() always called statRouteInfo() twice and rebuilt
active tablet ids every round. These full-route/stat rebuilds allocate
many temporary maps/sets, increasing young GC frequency.

Fix:
- Run the second statRouteInfo() only when smooth-upgrade migration
actually moved tablets.
- Add a starvation breaker: after N consecutive active-unbalanced
rounds, force one INACTIVE_ONLY round.
- Refresh active tablet snapshot by configurable interval (default 60s)
instead of every round.

Config:
- cloud_active_tablet_ids_refresh_interval_second (default 60)
- cloud_active_unbalanced_force_inactive_after_rounds (default 10)

Test:
- Add unit tests for forced inactive trigger, active snapshot refresh
interval, and migrate no-op path.

Author: deardeng

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -3497,6 +3497,21 @@ public static int metaServiceRpcRetryTimes() {
                     + "Default 10000. <=0 disables TopN segmentation."})
     public static int cloud_active_partition_scheduling_topn = 10000;
 
+    @ConfField(mutable = true, masterOnly = true, description = {
+            "活跃 tablet 优先调度开启时，active 集合刷新间隔（秒）。默认 60 秒，"
+                    + "表示 60 秒内复用同一批 active tablet，避免每轮重算。",
+            "Refresh interval in seconds for the active-tablet snapshot when active priority scheduling is enabled. "
+                    + "Default 60 seconds. Reuses the same active-tablet set within the interval."})
+    public static long cloud_active_tablet_ids_refresh_interval_second = 60L;
+
+    @ConfField(mutable = true, masterOnly = true, description = {
+            "活跃 tablet 优先调度开启时，若 active 阶段连续 N 轮未达均衡，"
+                    + "则强制执行一轮 inactive 阶段以避免长期饥饿。默认 10，<=0 表示关闭该强制机制。",
+            "When active priority scheduling is enabled and the active phase remains unbalanced for N consecutive "
+                    + "rounds, force one inactive phase round to avoid long-term starvation. "
+                    + "Default 10. <=0 disables this forced mechanism."})
+    public static int cloud_active_unbalanced_force_inactive_after_rounds = 10;
+
     @ConfField(mutable = true, masterOnly = false)
     public static String security_checker_class_name = "";
 

fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java

@@ -149,6 +149,8 @@ private enum ActiveSchedulePhase {
     private BalanceTypeEnum globalBalanceTypeEnum = BalanceTypeEnum.getCloudWarmUpForRebalanceTypeEnum();
 
     private Set<Long> activeTabletIds = new HashSet<>();
+    private long lastActiveTabletIdsRefreshMs = 0L;
+    private int consecutiveActiveUnbalancedRounds = 0;
 
     // cache for scheduling order in one daemon run (rebuilt in statRouteInfo)
     // table/partition active count is computed from activeTabletIds
@@ -510,7 +512,7 @@ protected void runAfterCatalogReady() {
 
         LOG.info("cloud tablet rebalance begin");
         long start = System.currentTimeMillis();
-        activeTabletIds = getActiveTabletIds();
+        refreshActiveTabletIdsIfNeeded();
         globalBalanceTypeEnum = BalanceTypeEnum.getCloudWarmUpForRebalanceTypeEnum();
 
         buildClusterToBackendMap();
@@ -519,8 +521,10 @@ protected void runAfterCatalogReady() {
         }
 
         statRouteInfo();
-        migrateTabletsForSmoothUpgrade();
-        statRouteInfo();
+        boolean migrated = migrateTabletsForSmoothUpgrade();
+        if (migrated) {
+            statRouteInfo();
+        }
 
         indexBalanced = true;
         tableBalanced = true;
@@ -564,18 +568,15 @@ private void buildClusterToBackendMap() {
         LOG.info("cluster to backends {}", clusterToBes);
     }
 
-    private void migrateTabletsForSmoothUpgrade() {
+    private boolean migrateTabletsForSmoothUpgrade() {
+        boolean migrated = false;
         Pair<Long, Long> pair;
-        while (!tabletsMigrateTasks.isEmpty()) {
-            try {
-                pair = tabletsMigrateTasks.take();
-                LOG.debug("begin tablets migration from be {} to be {}", pair.first, pair.second);
-                migrateTablets(pair.first, pair.second);
-            } catch (InterruptedException e) {
-                LOG.warn("migrate tablets failed", e);
-                throw new RuntimeException(e);
-            }
+        while ((pair = tabletsMigrateTasks.poll()) != null) {
+            LOG.debug("begin tablets migration from be {} to be {}", pair.first, pair.second);
+            migrateTablets(pair.first, pair.second);
+            migrated = true;
         }
+        return migrated;
     }
 
     private void performBalancing() {
@@ -585,6 +586,7 @@ private void performBalancing() {
         // to the same table or partition onto the same BE, while `partition` scheduling later requires these tablets
         // to be dispersed across different BEs, resulting in unnecessary scheduling.
         if (!Config.enable_cloud_active_tablet_priority_scheduling) {
+            consecutiveActiveUnbalancedRounds = 0;
             // Legacy scheduling: schedule the full set.
             if (Config.enable_cloud_partition_balance) {
                 balanceAllPartitionsByPhase(ActiveSchedulePhase.ALL);
@@ -621,10 +623,15 @@ private void performBalancing() {
                         activeIndexBalanced, activeTableBalanced, activeBalanced, clusterToBes.size());
             }
 
-            if (!activeBalanced) {
+            boolean forceInactivePhase = shouldForceInactivePhase(activeBalanced);
+            if (!activeBalanced && !forceInactivePhase) {
                 // Active objects are not balanced yet; skip phase2 to avoid diluting scheduling budget.
                 return;
             }
+            if (forceInactivePhase) {
+                LOG.info("active phase is still unbalanced for {} consecutive rounds, force run INACTIVE_ONLY once",
+                        Config.cloud_active_unbalanced_force_inactive_after_rounds);
+            }
 
             // Phase 2: inactive (all - active), then global if enabled and ready.
             boolean phase2IndexBalanced = true;
@@ -635,12 +642,30 @@ private void performBalancing() {
             if (Config.enable_cloud_table_balance && phase2IndexBalanced) {
                 phase2TableBalanced = balanceAllTablesByPhase(ActiveSchedulePhase.INACTIVE_ONLY);
             }
-            if (Config.enable_cloud_global_balance && phase2IndexBalanced && phase2TableBalanced) {
+            if (Config.enable_cloud_global_balance && activeBalanced && phase2IndexBalanced && phase2TableBalanced) {
                 globalBalance();
             }
         }
     }
 
+    private boolean shouldForceInactivePhase(boolean activeBalanced) {
+        if (activeBalanced) {
+            consecutiveActiveUnbalancedRounds = 0;
+            return false;
+        }
+        int forceAfterRounds = Config.cloud_active_unbalanced_force_inactive_after_rounds;
+        if (forceAfterRounds <= 0) {
+            consecutiveActiveUnbalancedRounds = 0;
+            return false;
+        }
+        consecutiveActiveUnbalancedRounds++;
+        if (consecutiveActiveUnbalancedRounds < forceAfterRounds) {
+            return false;
+        }
+        consecutiveActiveUnbalancedRounds = 0;
+        return true;
+    }
+
     private boolean balanceAllPartitionsByPhase(ActiveSchedulePhase phase) {
         // Reuse existing "balanced" flags as a per-phase signal.
         indexBalanced = true;
@@ -1682,7 +1707,7 @@ private long findSourceBackend(List<Long> bes, Map<Long, Set<Tablet>> beToTablet
                     maxTabletsNum = tabletNum;
                 }
             } else {
-                LOG.info("backend {} not found", be);
+                LOG.debug("backend {} not found", be);
             }
         }
         return srcBe;
@@ -1939,6 +1964,21 @@ private Set<Long> getActiveTabletIds() {
         }
     }
 
+    private void refreshActiveTabletIdsIfNeeded() {
+        long nowMs = System.currentTimeMillis();
+        if (!shouldRefreshActiveTabletIds(nowMs)) {
+            return;
+        }
+        activeTabletIds = getActiveTabletIds();
+        lastActiveTabletIdsRefreshMs = nowMs;
+    }
+
+    private boolean shouldRefreshActiveTabletIds(long nowMs) {
+        long refreshIntervalSeconds = Math.max(1L, Config.cloud_active_tablet_ids_refresh_interval_second);
+        long refreshIntervalMs = TimeUnit.SECONDS.toMillis(refreshIntervalSeconds);
+        return lastActiveTabletIdsRefreshMs <= 0L || nowMs - lastActiveTabletIdsRefreshMs >= refreshIntervalMs;
+    }
+
     // Choose non-active (cold) tablet first to re-balance, to reduce impact on hot tablets.
     // Fallback to active/random if no cold tablet is available.
     private Tablet pickTabletPreferCold(long srcBe, Set<Tablet> tablets, Set<Long> activeTabletIds,

fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudTabletRebalancerTest.java

@@ -43,16 +43,22 @@
 public class CloudTabletRebalancerTest {
 
     private boolean oldEnableActiveScheduling;
+    private long oldActiveTabletIdsRefreshIntervalSecond;
+    private int oldForceInactiveAfterRounds;
 
     @BeforeEach
     public void setUp() {
         oldEnableActiveScheduling = Config.enable_cloud_active_tablet_priority_scheduling;
+        oldActiveTabletIdsRefreshIntervalSecond = Config.cloud_active_tablet_ids_refresh_interval_second;
+        oldForceInactiveAfterRounds = Config.cloud_active_unbalanced_force_inactive_after_rounds;
         Config.enable_cloud_active_tablet_priority_scheduling = true;
     }
 
     @AfterEach
     public void tearDown() {
         Config.enable_cloud_active_tablet_priority_scheduling = oldEnableActiveScheduling;
+        Config.cloud_active_tablet_ids_refresh_interval_second = oldActiveTabletIdsRefreshIntervalSecond;
+        Config.cloud_active_unbalanced_force_inactive_after_rounds = oldForceInactiveAfterRounds;
     }
 
     private static class TestRebalancer extends CloudTabletRebalancer {
@@ -79,6 +85,13 @@ private static void setField(Object obj, String name, Object value) throws Excep
         f.set(obj, value);
     }
 
+    @SuppressWarnings("unchecked")
+    private static <T> T getField(Object obj, String name) throws Exception {
+        Field f = CloudTabletRebalancer.class.getDeclaredField(name);
+        f.setAccessible(true);
+        return (T) f.get(obj);
+    }
+
     @SuppressWarnings("unchecked")
     private static <T> T invokePrivate(Object obj, String method, Class<?>[] types, Object[] args) throws Exception {
         Method m = CloudTabletRebalancer.class.getDeclaredMethod(method, types);
@@ -236,6 +249,63 @@ public void testPartitionEntryComparator_internalDbLastAndIdDescTieBreak() throw
         Assertions.assertEquals(200L, list.get(1).getKey());
         Assertions.assertEquals(100L, list.get(2).getKey(), "Internal db partition should be scheduled last");
     }
-}
 
+    @Test
+    public void testShouldForceInactivePhase_afterConsecutiveUnbalancedRounds() throws Exception {
+        TestRebalancer r = new TestRebalancer();
+        Config.cloud_active_unbalanced_force_inactive_after_rounds = 3;
+
+        boolean forceRound1 = invokePrivate(r, "shouldForceInactivePhase",
+                new Class<?>[] {boolean.class}, new Object[] {false});
+        boolean forceRound2 = invokePrivate(r, "shouldForceInactivePhase",
+                new Class<?>[] {boolean.class}, new Object[] {false});
+        boolean forceRound3 = invokePrivate(r, "shouldForceInactivePhase",
+                new Class<?>[] {boolean.class}, new Object[] {false});
+
+        Assertions.assertFalse(forceRound1);
+        Assertions.assertFalse(forceRound2);
+        Assertions.assertTrue(forceRound3);
+        Assertions.assertEquals(0, (int) getField(r, "consecutiveActiveUnbalancedRounds"));
+
+        boolean forceAfterBalanced = invokePrivate(r, "shouldForceInactivePhase",
+                new Class<?>[] {boolean.class}, new Object[] {true});
+        Assertions.assertFalse(forceAfterBalanced);
+        Assertions.assertEquals(0, (int) getField(r, "consecutiveActiveUnbalancedRounds"));
+    }
+
+    @Test
+    public void testShouldRefreshActiveTabletIds_respectsIntervalAndClamp() throws Exception {
+        TestRebalancer r = new TestRebalancer();
+
+        Config.cloud_active_tablet_ids_refresh_interval_second = 60L;
+        setField(r, "lastActiveTabletIdsRefreshMs", 0L);
+        boolean firstRound = invokePrivate(r, "shouldRefreshActiveTabletIds",
+                new Class<?>[] {long.class}, new Object[] {1000L});
+        Assertions.assertTrue(firstRound);
+
+        setField(r, "lastActiveTabletIdsRefreshMs", 1000L);
+        boolean beforeInterval = invokePrivate(r, "shouldRefreshActiveTabletIds",
+                new Class<?>[] {long.class}, new Object[] {60000L});
+        boolean atInterval = invokePrivate(r, "shouldRefreshActiveTabletIds",
+                new Class<?>[] {long.class}, new Object[] {61000L});
+        Assertions.assertFalse(beforeInterval);
+        Assertions.assertTrue(atInterval);
+
+        Config.cloud_active_tablet_ids_refresh_interval_second = 0L; // clamp to 1s
+        setField(r, "lastActiveTabletIdsRefreshMs", 1000L);
+        boolean beforeClampInterval = invokePrivate(r, "shouldRefreshActiveTabletIds",
+                new Class<?>[] {long.class}, new Object[] {1500L});
+        boolean atClampInterval = invokePrivate(r, "shouldRefreshActiveTabletIds",
+                new Class<?>[] {long.class}, new Object[] {2000L});
+        Assertions.assertFalse(beforeClampInterval);
+        Assertions.assertTrue(atClampInterval);
+    }
+
+    @Test
+    public void testMigrateTabletsForSmoothUpgrade_emptyQueueReturnsFalse() throws Exception {
+        TestRebalancer r = new TestRebalancer();
+        boolean migrated = invokePrivate(r, "migrateTabletsForSmoothUpgrade", new Class<?>[] {}, new Object[] {});
+        Assertions.assertFalse(migrated);
+    }
+}