[fix](scheduler) disable query and load when BE is shutting down

Author: morningman

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

@@ -101,7 +101,7 @@ private boolean isColocated() {
     public long getColocatedBeId(String clusterId) throws ComputeGroupException {
         CloudSystemInfoService infoService = ((CloudSystemInfoService) Env.getCurrentSystemInfo());
         List<Backend> bes = infoService.getBackendsByClusterId(clusterId).stream()
-                .filter(be -> !be.isQueryDisabled()).collect(Collectors.toList());
+                .filter(be -> be.isQueryAvailable()).collect(Collectors.toList());
         String clusterName = infoService.getClusterNameByClusterId(clusterId);
         if (bes.isEmpty()) {
             LOG.warn("failed to get available be, cluster: {}-{}", clusterName, clusterId);

fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java

@@ -92,7 +92,7 @@ public StreamLoadHandler(TStreamLoadPutRequest request, AtomicInteger indexId,
     public static Backend selectBackend(String clusterName) throws LoadException {
         List<Backend> backends = ((CloudSystemInfoService) Env.getCurrentSystemInfo())
                 .getBackendsByClusterName(clusterName)
-                .stream().filter(Backend::isAlive)
+                .stream().filter(Backend::isLoadAvailable)
                 .collect(Collectors.toList());
 
         if (backends.isEmpty()) {

fe/fe-core/src/main/java/org/apache/doris/system/Backend.java

@@ -293,20 +293,38 @@ public void setLastStreamLoadTime(long lastStreamLoadTime) {
         this.backendStatus.lastStreamLoadTime = lastStreamLoadTime;
     }
 
+    // ATTN: This method only the value of "isQueryDisabled",
+    // it does not determine the backend IS queryable or not, use isQueryAvailable instead.
     public boolean isQueryDisabled() {
         return backendStatus.isQueryDisabled;
     }
 
-    public void setQueryDisabled(boolean isQueryDisabled) {
-        this.backendStatus.isQueryDisabled = isQueryDisabled;
+    // return true if be status is changed
+    public boolean setQueryDisabled(boolean isQueryDisabled) {
+        if (this.backendStatus.isQueryDisabled != isQueryDisabled) {
+            this.backendStatus.isQueryDisabled = isQueryDisabled;
+            return true;
+        }
+        return false;
     }
 
+    // ATTN: This method only the value of "isLoadDisabled",
+    // it does not determine the backend IS loadable or not, use isLoadAvailable instead.
     public boolean isLoadDisabled() {
         return backendStatus.isLoadDisabled;
     }
 
-    public void setLoadDisabled(boolean isLoadDisabled) {
-        this.backendStatus.isLoadDisabled = isLoadDisabled;
+    // return true if be status is changed
+    public boolean setLoadDisabled(boolean isLoadDisabled) {
+        if (this.backendStatus.isLoadDisabled != isLoadDisabled) {
+            this.backendStatus.isLoadDisabled = isLoadDisabled;
+            return true;
+        }
+        return false;
+    }
+
+    public boolean isShutDown() {
+        return isShutDown.get();
     }
 
     public void setActive(boolean isActive) {
@@ -524,15 +542,15 @@ public boolean isDecommissioning() {
     }
 
     public boolean isQueryAvailable() {
-        return isAlive() && !isQueryDisabled() && !isShutDown.get();
+        return isAlive() && !isQueryDisabled() && !isShutDown();
     }
 
     public boolean isScheduleAvailable() {
-        return isAlive() && !isDecommissioned();
+        return isAlive() && !isDecommissioned() && !isShutDown();
     }
 
     public boolean isLoadAvailable() {
-        return isAlive() && !isLoadDisabled();
+        return isAlive() && !isLoadDisabled() && !isShutDown();
     }
 
     public void setDisks(ImmutableMap<String, DiskInfo> disks) {

fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java

@@ -998,17 +998,11 @@ public void modifyBackends(ModifyBackendClause alterClause) throws UserException
             }
 
             if (alterClause.isQueryDisabled() != null) {
-                if (!alterClause.isQueryDisabled().equals(be.isQueryDisabled())) {
-                    be.setQueryDisabled(alterClause.isQueryDisabled());
-                    shouldModify = true;
-                }
+                shouldModify = be.setQueryDisabled(alterClause.isQueryDisabled());
             }
 
             if (alterClause.isLoadDisabled() != null) {
-                if (!alterClause.isLoadDisabled().equals(be.isLoadDisabled())) {
-                    be.setLoadDisabled(alterClause.isLoadDisabled());
-                    shouldModify = true;
-                }
+                shouldModify = be.setLoadDisabled(alterClause.isLoadDisabled());
             }
 
             if (shouldModify) {
@@ -1052,17 +1046,11 @@ public void modifyBackends(ModifyBackendOp op) throws UserException {
             }
 
             if (op.isQueryDisabled() != null) {
-                if (!op.isQueryDisabled().equals(be.isQueryDisabled())) {
-                    be.setQueryDisabled(op.isQueryDisabled());
-                    shouldModify = true;
-                }
+                shouldModify = be.setQueryDisabled(op.isQueryDisabled());
             }
 
             if (op.isLoadDisabled() != null) {
-                if (!op.isLoadDisabled().equals(be.isLoadDisabled())) {
-                    be.setLoadDisabled(op.isLoadDisabled());
-                    shouldModify = true;
-                }
+                shouldModify = be.setLoadDisabled(op.isLoadDisabled());
             }
 
             if (shouldModify) {