[enhance](cloud) Add a policy when be abnormal, tablet delay switch be

Author: deardeng

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

@@ -3030,6 +3030,12 @@ public static int metaServiceRpcRetryTimes() {
     @ConfField(mutable = true, description = {"存算分离模式下是否开启大事务提交，默认false"})
     public static boolean enable_cloud_txn_lazy_commit = false;
 
+    @ConfField(mutable = true, description = {"存算分离模式下，当tablet分布的be异常，是否立即映射tablet到新的be上，默认true"})
+    public static boolean enable_immediate_be_assign = true;
+
+    @ConfField(mutable = true, description = {"存算分离模式下，当tablet分布的be异常，临时映射(secondary) be的有效期，默认60s"})
+    public static int secondary_be_validity_seconds = 60 ;
+
     // ATTN: DONOT add any config not related to cloud mode here
     // ATTN: DONOT add any config not related to cloud mode here
     // ATTN: DONOT add any config not related to cloud mode here

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

@@ -20,7 +20,6 @@
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.Partition;
 import org.apache.doris.catalog.Replica;
-import org.apache.doris.catalog.Replica.ReplicaContext;
 import org.apache.doris.cloud.system.CloudSystemInfoService;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
@@ -60,11 +59,16 @@ public class CloudReplica extends Replica {
     private long indexId = -1;
     @SerializedName(value = "idx")
     private long idx = -1;
+    // no need serialize
+    private long secondaryBe = -1;
 
     private Random rand = new Random();
 
     private Map<String, List<Long>> memClusterToBackends = new ConcurrentHashMap<String, List<Long>>();
 
+    // badBeIP, badTime
+    private Map<String, Long> badBeAndBadTime = new ConcurrentHashMap<>();
+
     public CloudReplica() {
     }
 
@@ -216,20 +220,44 @@ private long getBackendIdImpl(String cluster) {
             long backendId = clusterToBackends.get(clusterId).get(0);
             Backend be = Env.getCurrentSystemInfo().getBackend(backendId);
             if (be != null && be.isQueryAvailable()) {
+                // be normal
                 if (LOG.isDebugEnabled()) {
                     LOG.debug("backendId={} ", backendId);
                 }
+                badBeAndBadTime.remove(be.getHost());
+                secondaryBe = -1;
                 return backendId;
             }
+            // be abnormal
+            if (Config.enable_immediate_be_assign) {
+                // rehash immediate
+                return hashReplicaToBe(clusterId, false, true);
+            }
+            // be abnormal but use secondary
+            if (be != null) {
+                if (!badBeAndBadTime.containsKey(be.getHost())) {
+                    badBeAndBadTime.put(be.getHost(), System.currentTimeMillis());
+                }
+                if (secondaryBe == -1) {
+                    secondaryBe = hashReplicaToBe(clusterId, false, false);
+                }
+                if (System.currentTimeMillis() - badBeAndBadTime.get(be.getHost())
+                        > Config.secondary_be_validity_seconds * 1000L) {
+                    // set secondary be to tablet after secondary_be_validity_seconds
+                    setBeToTablet(clusterId, secondaryBe);
+                    secondaryBe = -1;
+                }
+                return secondaryBe;
+            }
         }
         if (DebugPointUtil.isEnable("CloudReplica.getBackendIdImpl.clusterToBackends")) {
             LOG.info("Debug Point enable CloudReplica.getBackendIdImpl.clusterToBackends");
             return -1;
         }
-        return hashReplicaToBe(clusterId, false);
+        return hashReplicaToBe(clusterId, false, true);
     }
 
-    public long hashReplicaToBe(String clusterId, boolean isBackGround) {
+    public long hashReplicaToBe(String clusterId, boolean isBackGround, boolean setToTablet) {
         // TODO(luwei) list should be sorted
         List<Backend> clusterBes = ((CloudSystemInfoService) Env.getCurrentSystemInfo())
                 .getBackendsByClusterId(clusterId);
@@ -270,12 +298,18 @@ public long hashReplicaToBe(String clusterId, boolean isBackGround) {
                 pickedBeId, getId(), partitionId, availableBes.size(), idx, index,
                 hashCode == null ? -1 : hashCode.asLong());
 
+        if (setToTablet) {
+            setBeToTablet(clusterId, pickedBeId);
+        }
+
+        return pickedBeId;
+    }
+
+    private void setBeToTablet(String clusterId, long pickedBeId) {
         // save to clusterToBackends map
         List<Long> bes = new ArrayList<Long>();
         bes.add(pickedBeId);
         clusterToBackends.put(clusterId, bes);
-
-        return pickedBeId;
     }
 
     public List<Long> hashReplicaToBes(String clusterId, boolean isBackGround, int replicaNum) {

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

@@ -502,7 +502,7 @@ private void completeRouteInfo() {
                     Map<String, List<Long>> clusterToBackends =
                             ((CloudReplica) replica).getClusterToBackends();
                     if (!clusterToBackends.containsKey(cluster)) {
-                        long beId = ((CloudReplica) replica).hashReplicaToBe(cluster, true);
+                        long beId = ((CloudReplica) replica).hashReplicaToBe(cluster, true, true);
                         if (beId <= 0) {
                             assignedErrNum++;
                             continue;