branch-4.1: [fix](fe) Skip decommissioning BE for load

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

@@ -54,7 +54,8 @@ protected List<Long> getAvailableBackendIds(long jobId) throws LoadException {
         return ((CloudSystemInfoService) Env.getCurrentSystemInfo())
                 .getBackendsByClusterName(routineLoadJob.getCloudCluster())
                 .stream()
-                .filter(Backend::isAlive)
+                .filter(Backend::isLoadAvailable)
+                .filter(backend -> !backend.isDecommissioned() && !backend.isDecommissioning())
                 .map(Backend::getId)
                 .collect(Collectors.toList());
     }

fe/fe-core/src/main/java/org/apache/doris/datasource/kafka/KafkaUtil.java

@@ -238,8 +238,7 @@ private static InternalService.PProxyResult getInfoRequest(InternalService.PProx
                 List<Long> backendIds = new ArrayList<>();
                 for (Long beId : Env.getCurrentSystemInfo().getAllBackendIds(true)) {
                     Backend backend = Env.getCurrentSystemInfo().getBackend(beId);
-                    if (backend != null && backend.isLoadAvailable()
-                            && !backend.isDecommissioned()
+                    if (isBackendAvailableForMetaRequest(backend)
                             && !failedBeIds.contains(beId)
                             && !Env.getCurrentEnv().getRoutineLoadManager().isInBlacklist(beId)) {
                         backendIds.add(beId);
@@ -255,9 +254,9 @@ private static InternalService.PProxyResult getInfoRequest(InternalService.PProx
                     Map<Long, Long> blacklist = Env.getCurrentEnv().getRoutineLoadManager().getBlacklist();
                     for (Long beId : blacklist.keySet()) {
                         Backend backend = Env.getCurrentSystemInfo().getBackend(beId);
-                        if (backend != null) {
+                        if (isBackendAvailableForMetaRequest(backend) && !failedBeIds.contains(beId)) {
                             backendIds.add(beId);
-                        } else {
+                        } else if (backend == null) {
                             blacklist.remove(beId);
                             LOG.warn("remove stale backend {} from routine load blacklist when getting kafka meta",
                                     beId);
@@ -329,4 +328,9 @@ private static InternalService.PProxyResult getInfoRequest(InternalService.PProx
             MetricRepo.COUNTER_ROUTINE_LOAD_GET_META_COUNT.increase(1L);
         }
     }
+
+    private static boolean isBackendAvailableForMetaRequest(Backend backend) {
+        return backend != null && backend.isLoadAvailable()
+                && !backend.isDecommissioned() && !backend.isDecommissioning();
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java

@@ -284,6 +284,7 @@ public static Backend selectBackend(String cloudCluster, long preferredBackendId
                     .getBackendsByClusterName(cloudCluster)
                     .stream()
                     .filter(Backend::isLoadAvailable)
+                    .filter(backend -> !backend.isDecommissioned() && !backend.isDecommissioning())
                     .collect(Collectors.toList());
             if (bes.isEmpty()) {
                 throw new JobException(SystemInfoService.NO_BACKEND_LOAD_AVAILABLE_MSG

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

@@ -336,7 +336,8 @@ private long selectBackendForLocalGroupCommitInternal(long tableId) throws LoadE
         }
         List<String> backendsInfo = backends.stream()
                 .map(be -> "{ beId=" + be.getId() + ", alive=" + be.isAlive() + ", active=" + be.isActive()
-                        + ", decommission=" + be.isDecommissioned() + " }")
+                        + ", decommission=" + be.isDecommissioned()
+                        + ", decommissioning=" + be.isDecommissioning() + " }")
                 .collect(Collectors.toList());
         throw new LoadException("No suitable backend " + ", backends = " + backendsInfo);
     }
@@ -370,16 +371,13 @@ private Long getCachedBackend(String cluster, long tableId) {
     }
 
     private boolean isBackendAvailable(Backend backend, String cluster) {
-        if (backend == null || !backend.isAlive() || backend.isDecommissioned() || !backend.isLoadAvailable()) {
+        if (backend == null || !backend.isAlive() || backend.isDecommissioned() || backend.isDecommissioning()
+                || !backend.isLoadAvailable()) {
             return false;
         }
         if (!Config.isCloudMode()) {
             return true;
         }
-        // for cloud mode
-        if (backend.isDecommissioning()) {
-            return false;
-        }
         return cluster == null || cluster.equals(backend.getCloudClusterName());
     }
 

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

@@ -92,7 +92,8 @@ 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::isLoadAvailable)
+                .stream().filter(backend -> backend.isLoadAvailable() && !backend.isDecommissioned()
+                        && !backend.isDecommissioning())
                 .collect(Collectors.toList());
 
         if (backends.isEmpty()) {

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

@@ -144,8 +144,13 @@ public void removeMultiLoadTaskTxnIdToRoutineLoadJobId(long txnId) {
     }
 
     public void updateBeIdToMaxConcurrentTasks() {
-        beIdToMaxConcurrentTasks = Env.getCurrentSystemInfo().getAllBackendIds(true).stream().collect(
-                Collectors.toMap(beId -> beId, beId -> Config.max_routine_load_task_num_per_be));
+        beIdToMaxConcurrentTasks = Env.getCurrentSystemInfo().getAllBackendIds(true).stream()
+                .filter(beId -> {
+                    Backend backend = Env.getCurrentSystemInfo().getBackend(beId);
+                    return backend != null && backend.isLoadAvailable()
+                            && !backend.isDecommissioned() && !backend.isDecommissioning();
+                })
+                .collect(Collectors.toMap(beId -> beId, beId -> Config.max_routine_load_task_num_per_be));
     }
 
     // this is not real-time number
@@ -514,13 +519,15 @@ public long getAvailableBeForTask(long jobId, long previousBeId) throws UserExce
             updateBeIdToMaxConcurrentTasks();
             Map<Long, Integer> beIdToConcurrentTasks = getBeCurrentTasksNumMap();
             int previousBeIdleTaskNum = 0;
+            boolean previousBeAvailable = false;
 
             // 1. Find if the given BE id has more than half of available slots
             if (previousBeId != -1L && availableBeIds.contains(previousBeId)) {
                 // get the previousBackend info
                 Backend previousBackend = Env.getCurrentSystemInfo().getBackend(previousBeId);
                 // check previousBackend is not null && load available
                 if (previousBackend != null && previousBackend.isLoadAvailable()) {
+                    previousBeAvailable = true;
                     if (!beIdToMaxConcurrentTasks.containsKey(previousBeId)) {
                         previousBeIdleTaskNum = 0;
                     } else if (beIdToConcurrentTasks.containsKey(previousBeId)) {
@@ -529,7 +536,8 @@ public long getAvailableBeForTask(long jobId, long previousBeId) throws UserExce
                     } else {
                         previousBeIdleTaskNum = beIdToMaxConcurrentTasks.get(previousBeId);
                     }
-                    if (previousBeIdleTaskNum == Config.max_routine_load_task_num_per_be) {
+                    if (previousBeIdleTaskNum > 0
+                            && previousBeIdleTaskNum == Config.max_routine_load_task_num_per_be) {
                         return previousBeId;
                     }
                 }
@@ -558,7 +566,7 @@ public long getAvailableBeForTask(long jobId, long previousBeId) throws UserExce
             }
             // 4. on the basis of selecting the maximum idle slot be,
             //    try to reuse the object cache as much as possible
-            if (previousBeIdleTaskNum == maxIdleSlotNum) {
+            if (previousBeAvailable && previousBeIdleTaskNum > 0 && previousBeIdleTaskNum == maxIdleSlotNum) {
                 return previousBeId;
             }
             return resultBeId;

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

@@ -42,6 +42,7 @@ public class BeSelectionPolicy {
     public boolean needScheduleAvailable = false;
     public boolean needQueryAvailable = false;
     public boolean needLoadAvailable = false;
+    public boolean needNonDecommissioned = false;
     // Resource tag. Empty means no need to consider resource tag.
     public Set<Tag> resourceTags = Sets.newHashSet();
     // storage medium. null means no need to consider storage medium.
@@ -86,6 +87,12 @@ public Builder needQueryAvailable() {
 
         public Builder needLoadAvailable() {
             policy.needLoadAvailable = true;
+            policy.needNonDecommissioned = true;
+            return this;
+        }
+
+        public Builder needNonDecommissioned() {
+            policy.needNonDecommissioned = true;
             return this;
         }
 
@@ -156,6 +163,7 @@ private boolean isMatch(Backend backend) {
         if (needScheduleAvailable && !backend.isScheduleAvailable()
                 || needQueryAvailable && !backend.isQueryAvailable()
                 || needLoadAvailable && !backend.isLoadAvailable()
+                || needNonDecommissioned && (backend.isDecommissioned() || backend.isDecommissioning())
                 || (!resourceTags.isEmpty() && !resourceTags.contains(backend.getLocationTag()))
                 || storageMedium != null && !backend.hasSpecifiedStorageMedium(storageMedium)
                 || (requireAliveBe && !backend.isAlive())) {
@@ -231,8 +239,10 @@ public List<Backend> getCandidateBackends(Collection<Backend> backends) {
 
     @Override
     public String toString() {
-        return String.format("computeNode=%s | query=%s | load=%s | schedule=%s | tags=%s | medium=%s",
+        return String.format("computeNode=%s | query=%s | load=%s | schedule=%s | nonDecommissioned=%s"
+                        + " | tags=%s | medium=%s",
                 preferComputeNode, needQueryAvailable, needLoadAvailable, needScheduleAvailable,
+                needNonDecommissioned,
                 resourceTags.stream().map(tag -> tag.toString()).collect(Collectors.joining(",")), storageMedium);
     }
 }

fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java

@@ -0,0 +1,69 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.load;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class StreamLoadHandlerTest {
+    @Test
+    public void testSelectBackendSkipsDecommissioningBackend() throws Exception {
+        SystemInfoService originalSystemInfoService = Env.getCurrentSystemInfo();
+        Backend decommissioningBackend = createBackend(10001L, "127.0.0.1");
+        decommissioningBackend.setDecommissioning(true);
+        Backend selectedBackend = createBackend(10002L, "127.0.0.2");
+        CloudSystemInfoService systemInfoService =
+                new TestCloudSystemInfoService(Arrays.asList(decommissioningBackend, selectedBackend));
+
+        try {
+            Deencapsulation.setField(Env.getCurrentEnv(), "systemInfo", systemInfoService);
+
+            Assert.assertEquals(selectedBackend.getId(), StreamLoadHandler.selectBackend("cluster0").getId());
+        } finally {
+            Deencapsulation.setField(Env.getCurrentEnv(), "systemInfo", originalSystemInfoService);
+        }
+    }
+
+    private Backend createBackend(long id, String host) {
+        Backend backend = new Backend(id, host, 9050);
+        backend.setAlive(true);
+        return backend;
+    }
+
+    private static class TestCloudSystemInfoService extends CloudSystemInfoService {
+        private final List<Backend> backends;
+
+        private TestCloudSystemInfoService(List<Backend> backends) {
+            this.backends = backends;
+        }
+
+        @Override
+        public List<Backend> getBackendsByClusterName(final String clusterName) {
+            return backends;
+        }
+    }
+}