diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java index 126b29998eba5f..c954c9ab1bc364 100644 --- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java +++ b/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; diff --git a/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java b/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java index 3eb001ba624be2..8e475bc07dc049 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java +++ b/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) { } } - diff --git a/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java index 5c0ce52361e869..e96e349eb564cc 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java +++ b/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 cachedResourceGroupSuccQuorum = Collections.emptyMap(); private final long dbId; @@ -473,6 +480,9 @@ private void checkCommitStatus(List tableList, TransactionState transacti TabletInvertedIndex tabletInvertedIndex = env.getTabletInvertedIndex(); Map> tabletToBackends = new HashMap<>(); Map idToTable = new HashMap<>(); + Map resourceGroupSuccQuorum = getResourceGroupSuccQuorum(); + Map 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
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
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
tableList, TransactionState transacti throw new TabletQuorumFailedException(transactionId, errMsg); } + + for (Entry 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 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 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; } }