[feature](fe) Add resource group success quorum check (#66751)

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

@@ -680,6 +680,10 @@ public class Config extends ConfigBase {
             "Minimal number of write successful replicas for load job."})
     public static short min_load_replica_num = -1;
 
+    @ConfField(mutable = true, masterOnly = true, description = "Minimum number of successfully written replicas "
+            + "required in each resource group for a load job.")
+    public static volatile String[] resource_group_load_success_quorum = {};
+
     @ConfField(description = {"load job 调度器的执行间隔，单位是秒。",
             "The interval of load job scheduler, in seconds."})
     public static int load_checker_interval_second = 5;

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

@@ -126,7 +126,7 @@ public class Backend implements Writable {
     // the locationTag is also saved in tagMap, use a single field here to avoid
     // creating this everytime we get it.
     @SerializedName(value = "locationTag", alternate = {"tag"})
-    private Tag locationTag = Tag.DEFAULT_BACKEND_TAG;
+    private volatile Tag locationTag = Tag.DEFAULT_BACKEND_TAG;
 
     @SerializedName("nodeRole")
     private Tag nodeRoleTag = Tag.DEFAULT_NODE_ROLE_TAG;
@@ -1138,4 +1138,3 @@ public static Backend fromThrift(TBackend backend) {
     }
 
 }
-

fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java

@@ -28,6 +28,7 @@
 import org.apache.doris.catalog.Partition.PartitionState;
 import org.apache.doris.catalog.PartitionInfo;
 import org.apache.doris.catalog.Replica;
+import org.apache.doris.catalog.ReplicaAllocation;
 import org.apache.doris.catalog.Table;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.catalog.Tablet;
@@ -56,7 +57,9 @@
 import org.apache.doris.persist.CleanLabelOperationLog;
 import org.apache.doris.persist.EditLog;
 import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.resource.Tag;
 import org.apache.doris.statistics.AnalysisManager;
+import org.apache.doris.system.Backend;
 import org.apache.doris.task.AgentBatchTask;
 import org.apache.doris.task.AgentTaskExecutor;
 import org.apache.doris.task.ClearTransactionTask;
@@ -81,6 +84,7 @@
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -114,6 +118,9 @@ private enum PublishResult {
     // the max number of txn that can be remove per round.
     // set it to avoid holding lock too long when removing too many txns per round.
     private static final int MAX_REMOVE_TXN_PER_ROUND = 10000;
+    // ConfigBase replaces the array on every update, so its identity is the cache version.
+    private static volatile String[] cachedResourceGroupSuccQuorumConfig;
+    private static volatile Map<String, Integer> cachedResourceGroupSuccQuorum = Collections.emptyMap();
 
     private final long dbId;
 
@@ -473,6 +480,9 @@ private void checkCommitStatus(List<Table> tableList, TransactionState transacti
         TabletInvertedIndex tabletInvertedIndex = env.getTabletInvertedIndex();
         Map<Long, Set<Long>> tabletToBackends = new HashMap<>();
         Map<Long, Table> idToTable = new HashMap<>();
+        Map<String, Integer> resourceGroupSuccQuorum = getResourceGroupSuccQuorum();
+        Map<Long, String> backendLocationTags = resourceGroupSuccQuorum.isEmpty()
+                ? Collections.emptyMap() : new HashMap<>();
         for (int i = 0; i < tableList.size(); i++) {
             idToTable.put(tableList.get(i).getId(), tableList.get(i));
         }
@@ -589,6 +599,8 @@ private void checkCommitStatus(List<Table> tableList, TransactionState transacti
 
                 // (TODO): ignore the alter index if txn id is less than sc sched watermark
                 int loadRequiredReplicaNum = table.getLoadRequiredReplicaNum(partition.getId());
+                ReplicaAllocation replicaAllocation = resourceGroupSuccQuorum.isEmpty() ? null
+                        : table.getPartitionInfo().getReplicaAllocation(partition.getId());
                 for (MaterializedIndex index : allIndices) {
                     for (Tablet tablet : index.getTablets()) {
                         tabletSuccReplicas.clear();
@@ -606,6 +618,12 @@ private void checkCommitStatus(List<Table> tableList, TransactionState transacti
                                 throw new TransactionCommitFailedException("could not find replica for tablet ["
                                         + tabletId + "], backend [" + tabletBackend + "]");
                             }
+                            if (!resourceGroupSuccQuorum.isEmpty()) {
+                                backendLocationTags.computeIfAbsent(tabletBackend, backendId -> {
+                                    Backend backend = env.getCurrentSystemInfo().getBackend(backendId);
+                                    return backend == null ? "" : backend.getLocationTag().value;
+                                });
+                            }
 
                             // if the tablet have no replica's to commit or the tablet is a rolling up tablet,
                             // the commit backends maybe null
@@ -649,9 +667,71 @@ private void checkCommitStatus(List<Table> tableList, TransactionState transacti
 
                             throw new TabletQuorumFailedException(transactionId, errMsg);
                         }
+
+                        for (Entry<String, Integer> entry : resourceGroupSuccQuorum.entrySet()) {
+                            String resourceGroup = entry.getKey();
+                            int replicaNumInResourceGroup = replicaAllocation.getReplicaNumByTag(
+                                    Tag.createNotCheck(Tag.TYPE_LOCATION, resourceGroup));
+                            int requiredInResourceGroup = Math.min(entry.getValue(), replicaNumInResourceGroup);
+                            if (requiredInResourceGroup == 0) {
+                                continue;
+                            }
+
+                            int succInResourceGroup = 0;
+                            for (Replica replica : tabletSuccReplicas) {
+                                if (resourceGroup.equals(
+                                        backendLocationTags.get(replica.getBackendIdWithoutException()))) {
+                                    succInResourceGroup++;
+                                }
+                            }
+                            if (succInResourceGroup < requiredInResourceGroup) {
+                                String writeDetail = getTabletWriteDetail(tabletSuccReplicas,
+                                        tabletWriteFailedReplicas, tabletVersionFailedReplicas);
+                                String errMsg = String.format("Failed to commit txn %s, cause tablet %s resource "
+                                                + "group success quorum failed for %s: required %s successful "
+                                                + "replicas, but only %s succeeded. table %s, partition: [ id=%s, "
+                                                + "commit version %s, visible version %s ], this tablet detail: %s. "
+                                                + "Please try again later.", transactionId, tablet.getId(),
+                                        resourceGroup, requiredInResourceGroup, succInResourceGroup, tableId,
+                                        partition.getId(), partition.getCommittedVersion(),
+                                        partition.getVisibleVersion(), writeDetail);
+                                LOG.info(errMsg);
+                                throw new TabletQuorumFailedException(transactionId, errMsg);
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    private static Map<String, Integer> getResourceGroupSuccQuorum() {
+        String[] config = Config.resource_group_load_success_quorum;
+        if (config == cachedResourceGroupSuccQuorumConfig) {
+            return cachedResourceGroupSuccQuorum;
+        }
+        synchronized (DatabaseTransactionMgr.class) {
+            config = Config.resource_group_load_success_quorum;
+            if (config == cachedResourceGroupSuccQuorumConfig) {
+                return cachedResourceGroupSuccQuorum;
+            }
+            Map<String, Integer> parsedConfig = new HashMap<>();
+            for (String item : config) {
+                String[] parts = item.split(":", -1);
+                try {
+                    int configuredMin = Integer.parseInt(parts.length == 2 ? parts[1].trim() : "");
+                    if (parts[0].trim().isEmpty() || configuredMin < 0) {
+                        throw new NumberFormatException();
                     }
+                    parsedConfig.put(parts[0].trim(), configuredMin);
+                } catch (NumberFormatException e) {
+                    LOG.warn("Invalid resource_group_load_success_quorum item '{}', ignored. Expected format "
+                            + "resource_group:min_success_replicas with a non-negative integer.", item);
                 }
             }
+            cachedResourceGroupSuccQuorum = parsedConfig;
+            cachedResourceGroupSuccQuorumConfig = config;
+            return parsedConfig;
         }
     }