[Fix](group commit) Fix multiple cluster group commit BE select strat…

…egy (apache#38644)

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

@@ -58,10 +58,10 @@ public class GroupCommitManager {
 
     private Set<Long> blockedTableIds = new HashSet<>();
 
-    // Table id to BE id map. Only for group commit.
-    private Map<Long, Long> tableToBeMap = new ConcurrentHashMap<>();
-    // BE id to pressure map. Only for group commit.
-    private Map<Long, SlidingWindowCounter> tablePressureMap = new ConcurrentHashMap<>();
+    // Encoded <Cluster and Table id> to BE id map. Only for group commit.
+    private final Map<String, Long> tableToBeMap = new ConcurrentHashMap<>();
+    // Table id to pressure map. Only for group commit.
+    private final Map<Long, SlidingWindowCounter> tableToPressureMap = new ConcurrentHashMap<>();
 
     public boolean isBlock(long tableId) {
         return blockedTableIds.contains(tableId);
@@ -243,13 +243,13 @@ public long selectBackendForGroupCommitInternal(long tableId, String cluster, bo
 
     private long selectBackendForCloudGroupCommitInternal(long tableId, String cluster)
             throws DdlException, LoadException {
-        LOG.debug("cloud group commit select be info, tableToBeMap {}, tablePressureMap {}", tableToBeMap.toString(),
-                tablePressureMap.toString());
+        LOG.debug("cloud group commit select be info, tableToBeMap {}, tablePressureMap {}",
+                tableToBeMap.toString(), tableToPressureMap.toString());
         if (Strings.isNullOrEmpty(cluster)) {
             ErrorReport.reportDdlException(ErrorCode.ERR_NO_CLUSTER_ERROR);
         }
 
-        Long cachedBackendId = getCachedBackend(tableId);
+        Long cachedBackendId = getCachedBackend(cluster, tableId);
         if (cachedBackendId != null) {
             return cachedBackendId;
         }
@@ -261,7 +261,7 @@ private long selectBackendForCloudGroupCommitInternal(long tableId, String clust
             throw new LoadException("No alive backend");
         }
         // If the cached backend is not active or decommissioned, select a random new backend.
-        Long randomBackendId = getRandomBackend(tableId, backends);
+        Long randomBackendId = getRandomBackend(cluster, tableId, backends);
         if (randomBackendId != null) {
             return randomBackendId;
         }
@@ -274,8 +274,8 @@ private long selectBackendForCloudGroupCommitInternal(long tableId, String clust
 
     private long selectBackendForLocalGroupCommitInternal(long tableId) throws LoadException {
         LOG.debug("group commit select be info, tableToBeMap {}, tablePressureMap {}", tableToBeMap.toString(),
-                tablePressureMap.toString());
-        Long cachedBackendId = getCachedBackend(tableId);
+                tableToPressureMap.toString());
+        Long cachedBackendId = getCachedBackend(null, tableId);
         if (cachedBackendId != null) {
             return cachedBackendId;
         }
@@ -293,7 +293,7 @@ private long selectBackendForLocalGroupCommitInternal(long tableId) throws LoadE
         }
 
         // If the cached backend is not active or decommissioned, select a random new backend.
-        Long randomBackendId = getRandomBackend(tableId, backends);
+        Long randomBackendId = getRandomBackend(null, tableId, backends);
         if (randomBackendId != null) {
             return randomBackendId;
         }
@@ -305,38 +305,56 @@ private long selectBackendForLocalGroupCommitInternal(long tableId) throws LoadE
     }
 
     @Nullable
-    private Long getCachedBackend(long tableId) {
+    private Long getCachedBackend(String cluster, long tableId) {
         OlapTable table = (OlapTable) Env.getCurrentEnv().getInternalCatalog().getTableByTableId(tableId);
-        if (tableToBeMap.containsKey(tableId)) {
-            if (tablePressureMap.get(tableId).get() < table.getGroupCommitDataBytes()) {
-                Backend backend = Env.getCurrentSystemInfo().getBackend(tableToBeMap.get(tableId));
+        if (tableToBeMap.containsKey(encode(cluster, tableId))) {
+            if (tableToPressureMap.get(tableId).get() < table.getGroupCommitDataBytes()) {
+                // There are multiple threads getting cached backends for the same table.
+                // Maybe one thread removes the tableId from the tableToBeMap.
+                // Another thread gets the same tableId but can not find this tableId.
+                // So another thread needs to get the random backend.
+                Long backendId = tableToBeMap.get(encode(cluster, tableId));
+                Backend backend;
+                if (backendId != null) {
+                    backend = Env.getCurrentSystemInfo().getBackend(backendId);
+                } else {
+                    return null;
+                }
                 if (backend.isActive() && !backend.isDecommissioned()) {
                     return backend.getId();
                 } else {
-                    tableToBeMap.remove(tableId);
+                    tableToBeMap.remove(encode(cluster, tableId));
                 }
             } else {
-                tableToBeMap.remove(tableId);
+                tableToBeMap.remove(encode(cluster, tableId));
             }
         }
         return null;
     }
 
     @Nullable
-    private Long getRandomBackend(long tableId, List<Backend> backends) {
+    private Long getRandomBackend(String cluster, long tableId, List<Backend> backends) {
         OlapTable table = (OlapTable) Env.getCurrentEnv().getInternalCatalog().getTableByTableId(tableId);
         Collections.shuffle(backends);
         for (Backend backend : backends) {
             if (backend.isActive() && !backend.isDecommissioned()) {
-                tableToBeMap.put(tableId, backend.getId());
-                tablePressureMap.put(tableId,
+                tableToBeMap.put(encode(cluster, tableId), backend.getId());
+                tableToPressureMap.put(tableId,
                         new SlidingWindowCounter(table.getGroupCommitIntervalMs() / 1000 + 1));
                 return backend.getId();
             }
         }
         return null;
     }
 
+    private String encode(String cluster, long tableId) {
+        if (cluster == null) {
+            return String.valueOf(tableId);
+        } else {
+            return cluster + tableId;
+        }
+    }
+
     public void updateLoadData(long tableId, long receiveData) {
         if (tableId == -1) {
             LOG.warn("invalid table id: " + tableId);
@@ -359,10 +377,10 @@ public void updateLoadData(long tableId, long receiveData) {
     }
 
     public void updateLoadDataInternal(long tableId, long receiveData) {
-        if (tablePressureMap.containsKey(tableId)) {
-            tablePressureMap.get(tableId).add(receiveData);
+        if (tableToPressureMap.containsKey(tableId)) {
+            tableToPressureMap.get(tableId).add(receiveData);
             LOG.info("Update load data for table{}, receiveData {}, tablePressureMap {}", tableId, receiveData,
-                    tablePressureMap.toString());
+                    tableToPressureMap.toString());
         } else {
             LOG.warn("can not find backend id: {}", tableId);
         }