[opt](cloud) Prioritize scheduling the most recently active tablets in cloud

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

@@ -3442,6 +3442,11 @@ public static int metaServiceRpcRetryTimes() {
         "Thread pool size for asynchronous warm-up RPC dispatch during cloud tablet rebalancing, default 4"})
     public static int cloud_warm_up_rpc_async_pool_size = 4;
 
+    @ConfField(masterOnly = true, description = {"云上tablet均衡时，是否开启活跃tablet优先调度策略，默认打开"
+            + "When tablets are being balanced in the cloud, "
+            + "is the active tablet priority scheduling strategy enabled?  (Default: Enabled)"})
+    public static boolean enable_cloud_active_tablet_priority_scheduling = true;
+
     @ConfField(mutable = true, masterOnly = false)
     public static String security_checker_class_name = "";
 

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

@@ -175,6 +175,14 @@ protected long getBackendIdValue() {
         return -1L;
     }
 
+    // just record read,write ops
+    public long getBackendIdAndRecordAccessInfo() throws UserException {
+        // Use async version to avoid blocking getBackendIdImpl which is called frequently
+        TabletAccessStats.getInstance().recordAccessAsync(getId());
+        return getBackendId();
+    }
+
+    // just for ut
     public void setBackendId(long backendId) {
         if (backendId != -1) {
             throw new UnsupportedOperationException("setBackendId is not supported in Replica");

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

@@ -247,7 +247,7 @@ private Multimap<Long, Long> getNormalReplicaBackendPathMapImpl(String beEndpoin
     // return map of (BE id -> path hash) of normal replicas
     // for load plan.
     public Multimap<Long, Long> getNormalReplicaBackendPathMap() throws UserException {
-        return getNormalReplicaBackendPathMapImpl(null, (rep, be) -> rep.getBackendId());
+        return getNormalReplicaBackendPathMapImpl(null, (rep, be) -> rep.getBackendIdAndRecordAccessInfo());
     }
 
     // for cloud mode without ConnectContext. use BE IP to find replica
@@ -281,7 +281,15 @@ public List<Replica> getQueryableReplicas(long visibleVersion, Map<Long, Set<Lon
                 continue;
             }
 
-            Set<Long> thisBeAlivePaths = backendAlivePathHashs.get(replica.getBackendIdWithoutException());
+            long beId = -1;
+            try {
+                beId = replica.getBackendIdAndRecordAccessInfo();
+            } catch (UserException e) {
+                if (LOG.isDebugEnabled()) {
+                    LOG.debug("getBackendIdWithoutException: ", e);
+                }
+            }
+            Set<Long> thisBeAlivePaths = backendAlivePathHashs.get(beId);
             ReplicaState state = replica.getState();
             // if thisBeAlivePaths contains pathHash = 0, it mean this be hadn't report disks state.
             // should ignore this case.