[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/datasource/kinesis/KinesisUtil.java

@@ -101,16 +101,23 @@ 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);
  }
  }
  if (backendIds.isEmpty()) {
- for (Long beId : Env.getCurrentEnv().getRoutineLoadManager().getBlacklist().keySet()) {
- backendIds.add(beId);
+ Map<Long, Long> blacklist = Env.getCurrentEnv().getRoutineLoadManager().getBlacklist();
+ for (Long beId : blacklist.keySet()) {
+ Backend backend = Env.getCurrentSystemInfo().getBackend(beId);
+ if (isBackendAvailableForMetaRequest(backend) && !failedBeIds.contains(beId)) {
+ backendIds.add(beId);
+ } else if (backend == null) {
+ blacklist.remove(beId);
+ LOG.warn("remove stale backend {} from routine load blacklist when getting kinesis meta",
+ beId);
+ }
  }
  }
  if (backendIds.isEmpty()) {
@@ -162,4 +169,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

@@ -279,6 +279,7 @@ public static Backend selectBackend(String cloudCluster) throws JobException {
  .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

@@ -97,7 +97,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

@@ -146,8 +146,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
@@ -519,13 +524,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)) {
@@ -534,7 +541,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;
  }
  }
@@ -563,7 +571,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;
+ }
+ }
+}