This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new d80d0eb1dfd [feature](fe) Add resource group success quorum check
(#66751) (#67229)
d80d0eb1dfd is described below
commit d80d0eb1dfd49e3e4af16c1192f2ab06463e6df3
Author: deardeng <[email protected]>
AuthorDate: Mon Aug 31 16:26:06 2026 +0800
[feature](fe) Add resource group success quorum check (#66751) (#67229)
pick from https://github.com/apache/doris/pull/66751
Enforce configured minimum successful load replicas for each resource
group during transaction commit.
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../main/java/org/apache/doris/common/Config.java | 4 ++
.../main/java/org/apache/doris/system/Backend.java | 3 +-
.../doris/transaction/DatabaseTransactionMgr.java | 80 ++++++++++++++++++++++
3 files changed, 85 insertions(+), 2 deletions(-)
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 814a10024fb..0474ad7ba4c 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 3eb001ba624..8e475bc07dc 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 class Backend implements Writable {
}
}
-
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 5c0ce52361e..e96e349eb56 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;
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.BatchRemoveTransactionsOperationV2;
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.io.IOException;
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 @@ public class DatabaseTransactionMgr {
// 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 @@ public class DatabaseTransactionMgr {
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 @@ public class DatabaseTransactionMgr {
// (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 @@ public class DatabaseTransactionMgr {
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 @@ public class DatabaseTransactionMgr {
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;
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
