[draft](tablet scheduler) fix higher priority tablet add failed due to pending queue full

fe/fe-core/src/main/java/org/apache/doris/clone/ColocateTableCheckerAndBalancer.java

@@ -561,14 +561,14 @@ private void matchGroups() {
                                     tabletCtx.setIsUniqKeyMergeOnWrite(isUniqKeyMergeOnWrite);
 
                                     AddResult res = tabletScheduler.addTablet(tabletCtx, false /* not force */);
-                                    if (res == AddResult.LIMIT_EXCEED || res == AddResult.DISABLED) {
+                                    if (res == AddResult.DISABLED) {
                                         // tablet in scheduler exceed limit, or scheduler is disabled,
                                         // skip this group and check next one.
                                         LOG.info("tablet scheduler return: {}. stop colocate table check", res.name());
                                         break OUT;
                                     } else if (res == AddResult.ADDED) {
                                         counter.addToSchedulerTabletNum++;
-                                    }  else {
+                                    }  else if (res == AddResult.ALREADY_IN) {
                                         counter.tabletInScheduler++;
                                     }
                                 }

fe/fe-core/src/main/java/org/apache/doris/clone/TabletChecker.java

@@ -404,7 +404,7 @@ private LoopControlStatus handlePartitionTablet(Database db, OlapTable tbl, Part
                 tabletCtx.setIsUniqKeyMergeOnWrite(isUniqKeyMergeOnWrite);
 
                 AddResult res = tabletScheduler.addTablet(tabletCtx, false /* not force */);
-                if (res == AddResult.LIMIT_EXCEED || res == AddResult.DISABLED) {
+                if (res == AddResult.DISABLED) {
                     LOG.info("tablet scheduler return: {}. stop tablet checker", res.name());
                     return LoopControlStatus.BREAK_OUT;
                 } else if (res == AddResult.ADDED) {

fe/fe-core/src/main/java/org/apache/doris/clone/TabletScheduler.java

@@ -69,6 +69,7 @@
 import com.google.common.collect.EvictingQueue;
 import com.google.common.collect.HashBasedTable;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.MinMaxPriorityQueue;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
@@ -81,7 +82,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.PriorityQueue;
 import java.util.Queue;
 import java.util.Set;
 
@@ -120,7 +120,7 @@ public class TabletScheduler extends MasterDaemon {
      *
      * pendingTablets, allTabletTypes, runningTablets and schedHistory are protected by 'synchronized'
      */
-    private PriorityQueue<TabletSchedCtx> pendingTablets = new PriorityQueue<>();
+    private MinMaxPriorityQueue<TabletSchedCtx> pendingTablets = MinMaxPriorityQueue.create();
     private Map<Long, TabletSchedCtx.Type> allTabletTypes = Maps.newHashMap();
     // contains all tabletCtxs which state are RUNNING
     private Map<Long, TabletSchedCtx> runningTablets = Maps.newHashMap();
@@ -277,9 +277,15 @@ public synchronized AddResult addTablet(TabletSchedCtx tablet, boolean force) {
         // if this is not a force add,
         // and number of scheduling tablets exceed the limit,
         // refuse to add.
-        if (!force && (pendingTablets.size() > Config.max_scheduling_tablets
-                || runningTablets.size() > Config.max_scheduling_tablets)) {
-            return AddResult.LIMIT_EXCEED;
+        if (!force && (pendingTablets.size() >= Config.max_scheduling_tablets
+                || runningTablets.size() >= Config.max_scheduling_tablets)) {
+            TabletSchedCtx lowestPriorityTablet = pendingTablets.peekLast();
+            if (lowestPriorityTablet == null || lowestPriorityTablet.compareTo(tablet) <= 0) {
+                return AddResult.LIMIT_EXCEED;
+            }
+            pendingTablets.pollLast();
+            finalizeTabletCtx(lowestPriorityTablet, TabletSchedCtx.State.CANCELLED, Status.UNRECOVERABLE,
+                    "envit lower priority sched tablet because pending queue is full");
         }
 
         if (!contains || tablet.getType() == TabletSchedCtx.Type.REPAIR) {
@@ -312,11 +318,12 @@ public synchronized void cancelRebalanceDisk(AdminCancelRebalanceDiskStmt stmt)
      * Iterate current tablets, change their priority to VERY_HIGH if necessary.
      */
     public synchronized void changeTabletsPriorityToVeryHigh(long dbId, long tblId, List<Long> partitionIds) {
-        PriorityQueue<TabletSchedCtx> newPendingTablets = new PriorityQueue<>();
+        MinMaxPriorityQueue<TabletSchedCtx> newPendingTablets = MinMaxPriorityQueue.create();
         for (TabletSchedCtx tabletCtx : pendingTablets) {
             if (tabletCtx.getDbId() == dbId && tabletCtx.getTblId() == tblId
                     && partitionIds.contains(tabletCtx.getPartitionId())) {
                 tabletCtx.setPriority(Priority.VERY_HIGH);
+                tabletCtx.setLastVisitedTime(1L);
             }
             newPendingTablets.add(tabletCtx);
         }
@@ -1750,7 +1757,7 @@ private synchronized List<TabletSchedCtx> getNextTabletCtxBatch() {
             slotNum = 1;
         }
         while (list.size() < Config.schedule_batch_size && slotNum > 0) {
-            TabletSchedCtx tablet = pendingTablets.poll();
+            TabletSchedCtx tablet = pendingTablets.pollFirst();
             if (tablet == null) {
                 // no more tablets
                 break;
@@ -1952,6 +1959,11 @@ public void handleRunningTablets() {
         });
     }
 
+    // only use for fe ut
+    public MinMaxPriorityQueue<TabletSchedCtx> getPendingTabletQueue() {
+        return pendingTablets;
+    }
+
     public List<List<String>> getPendingTabletsInfo(int limit) {
         return collectTabletCtx(getPendingTablets(limit));
     }

fe/fe-core/src/test/java/org/apache/doris/clone/TabletSchedCtxTest.java

@@ -17,12 +17,15 @@
 
 package org.apache.doris.clone;
 
+import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.Replica;
 import org.apache.doris.catalog.ReplicaAllocation;
 import org.apache.doris.clone.TabletSchedCtx.Priority;
 import org.apache.doris.clone.TabletSchedCtx.Type;
+import org.apache.doris.common.Config;
 
 import com.google.common.collect.Lists;
+import com.google.common.collect.MinMaxPriorityQueue;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -32,10 +35,40 @@
 
 public class TabletSchedCtxTest {
 
+    @Test
+    public void testAddTablet() {
+        List<TabletSchedCtx> tablets = Lists.newArrayList();
+        ReplicaAllocation replicaAlloc = ReplicaAllocation.DEFAULT_ALLOCATION;
+        for (long i = 0; i < 20; i++) {
+            tablets.add(new TabletSchedCtx(Type.REPAIR, 1, 2, 3, 4,
+                    i, replicaAlloc, i));
+            tablets.add(new TabletSchedCtx(Type.BALANCE, 1, 2, 3, 4,
+                    1000 + i, replicaAlloc, i));
+        }
+        Collections.shuffle(tablets);
+        Config.max_scheduling_tablets = 5;
+        TabletScheduler scheduler = Env.getCurrentEnv().getTabletScheduler();
+        for (TabletSchedCtx tablet : tablets) {
+            scheduler.addTablet(tablet, false);
+        }
+
+        MinMaxPriorityQueue<TabletSchedCtx> queue = scheduler.getPendingTabletQueue();
+        List<TabletSchedCtx> gotTablets = Lists.newArrayList();
+        while (!queue.isEmpty()) {
+            gotTablets.add(queue.pollFirst());
+        }
+        Assert.assertEquals(Config.max_scheduling_tablets, gotTablets.size());
+        for (int i = 0; i < gotTablets.size(); i++) {
+            TabletSchedCtx tablet = gotTablets.get(i);
+            Assert.assertEquals(Type.REPAIR, tablet.getType());
+            Assert.assertEquals((long) i, tablet.getCreateTime());
+        }
+    }
+
     @Test
     public void testPriorityCompare() {
         // equal priority, but info3's last visit time is earlier than info2 and info1, so info1 should ranks ahead
-        PriorityQueue<TabletSchedCtx> pendingTablets = new PriorityQueue<>();
+        MinMaxPriorityQueue<TabletSchedCtx> pendingTablets = MinMaxPriorityQueue.create();
         ReplicaAllocation replicaAlloc = ReplicaAllocation.DEFAULT_ALLOCATION;
         TabletSchedCtx ctx1 = new TabletSchedCtx(Type.REPAIR,
                 1, 2, 3, 4, 1000, replicaAlloc, System.currentTimeMillis());