This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 2bc391a853d [feature](fe) Add resource group success quorum check
(#66751)
2bc391a853d is described below
commit 2bc391a853d15b7185f3221a9a3e3bb1b61cb288
Author: deardeng <[email protected]>
AuthorDate: Wed Aug 26 11:40:10 2026 +0800
[feature](fe) Add resource group success quorum check (#66751)
Related PR: #66680
Problem Summary: Load transaction commits only enforced the ordinary
replica quorum and could therefore succeed without a configured minimum
number of successful replicas in each availability resource group. Add a
mutable FE configuration and enforce the per-resource group success
floor in the centralized transaction commit check. Clamp each resource
group requirement to the partition declared replica allocation so
backend liveness changes and transient extra replicas do not weaken or
inflate the commit requirement.
### Release note
Add the mutable FE configuration resource_group_succ_quorum to require a
minimum number of successful load replicas per availability zone.
---
.../main/java/org/apache/doris/common/Config.java | 5 +-
.../main/java/org/apache/doris/system/Backend.java | 3 +-
.../doris/transaction/DatabaseTransactionMgr.java | 78 +++++++
.../java/org/apache/doris/catalog/BackendTest.java | 6 +
.../transaction/DatabaseTransactionMgrTest.java | 233 +++++++++++++++++++++
.../test_resource_group_load_success_quorum.groovy | 119 +++++++++++
...oup_load_success_quorum_min_load_replica.groovy | 122 +++++++++++
7 files changed, 563 insertions(+), 3 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 b28c4e7398b..f01d40f1d99 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
@@ -20,7 +20,6 @@ package org.apache.doris.common;
import java.io.File;
public class Config extends ConfigBase {
-
@ConfField(description = "The path of the user-defined configuration file,
used to store fe_custom.conf. "
+ "Configurations in this file will override those in fe.conf")
public static String custom_config_dir = EnvUtils.getDorisHome() + "/conf";
@@ -567,6 +566,10 @@ public class Config extends ConfigBase {
+ "a 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 = "The interval of the 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 d403c88732e..a4fe6e6105d 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 cb18868d0b7..9ad29be502f 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
@@ -29,6 +29,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;
@@ -60,7 +61,9 @@ import org.apache.doris.persist.CleanLabelOperationLog;
import org.apache.doris.persist.EditLog;
import org.apache.doris.persist.OperationType;
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;
@@ -120,6 +123,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
= Map.of();
private final long dbId;
@@ -494,6 +500,8 @@ 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() ? Map.of() : new HashMap<>();
for (int i = 0; i < tableList.size(); i++) {
idToTable.put(tableList.get(i).getId(), tableList.get(i));
}
@@ -610,6 +618,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();
@@ -627,6 +637,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
@@ -670,9 +686,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;
}
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/BackendTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/BackendTest.java
index 1c5142b7625..a2f4b769ebd 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/catalog/BackendTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/BackendTest.java
@@ -33,6 +33,7 @@ import org.junit.Test;
import java.io.DataInputStream;
import java.io.DataOutputStream;
+import java.lang.reflect.Modifier;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
@@ -98,6 +99,11 @@ public class BackendTest {
Assert.assertTrue(backend.isAlive());
}
+ @Test
+ public void testLocationTagIsSafelyPublished() throws NoSuchFieldException
{
+
Assert.assertTrue(Modifier.isVolatile(Backend.class.getDeclaredField("locationTag").getModifiers()));
+ }
+
@Test
public void diskInfoTest() {
Map<String, TDisk> diskInfos = new HashMap<String, TDisk>();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/transaction/DatabaseTransactionMgrTest.java
b/fe/fe-core/src/test/java/org/apache/doris/transaction/DatabaseTransactionMgrTest.java
index 7de6aa28aea..bc85c2134bf 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/transaction/DatabaseTransactionMgrTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/transaction/DatabaseTransactionMgrTest.java
@@ -22,8 +22,12 @@ import org.apache.doris.catalog.CatalogTestUtil;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.FakeEditLog;
import org.apache.doris.catalog.FakeEnv;
+import org.apache.doris.catalog.LocalReplica;
import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.Replica;
+import org.apache.doris.catalog.ReplicaAllocation;
import org.apache.doris.catalog.Table;
+import org.apache.doris.catalog.Tablet;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.FeMetaVersion;
@@ -31,14 +35,19 @@ import org.apache.doris.common.Pair;
import org.apache.doris.common.UserException;
import org.apache.doris.common.util.TimeUtils;
import org.apache.doris.meta.MetaContext;
+import org.apache.doris.mysql.authenticate.TestLogAppender;
+import org.apache.doris.resource.Tag;
+import org.apache.doris.system.Backend;
import org.apache.doris.task.PublishVersionTask;
import org.apache.doris.thrift.TPartitionVersionInfo;
import
org.apache.doris.transaction.GlobalTransactionMgrTest.SubTransactionInfo;
import org.apache.doris.transaction.TransactionState.LoadJobSourceType;
import org.apache.doris.tso.TSOService;
+import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
+import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.junit.After;
@@ -234,6 +243,230 @@ public class DatabaseTransactionMgrTest {
Assert.assertEquals(TransactionStatus.PREPARE,
transactionState2.getTransactionStatus());
}
+ @Test
+ public void testResourceGroupSuccessQuorum() throws UserException {
+ FakeEnv.setEnv(masterEnv);
+ String[] originalResourceGroupSuccQuorum =
Config.resource_group_load_success_quorum;
+ Backend backend1 =
masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId1);
+ Backend backend2 =
masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId2);
+ Backend backend3 =
masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId3);
+ Map<String, String> backend1TagMap =
ImmutableMap.copyOf(backend1.getTagMap());
+ Map<String, String> backend2TagMap =
ImmutableMap.copyOf(backend2.getTagMap());
+ Map<String, String> backend3TagMap =
ImmutableMap.copyOf(backend3.getTagMap());
+ backend1.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "group1"));
+ backend2.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "group1"));
+ backend3.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "group2"));
+ OlapTable table = (OlapTable)
masterEnv.getInternalCatalog().getDbOrMetaException(CatalogTestUtil.testDbId1)
+ .getTableOrMetaException(CatalogTestUtil.testTableId1);
+ ReplicaAllocation originalAllocation = table.getPartitionInfo()
+ .getReplicaAllocation(CatalogTestUtil.testPartitionId1);
+
table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1,
+ new ReplicaAllocation(ImmutableMap.of(
+ Tag.createNotCheck(Tag.TYPE_LOCATION, "group1"),
(short) 2,
+ Tag.createNotCheck(Tag.TYPE_LOCATION, "group2"),
(short) 1)));
+
+ try {
+ Config.resource_group_load_success_quorum = new String[]
{"group1:2", "group2:1"};
+ long transactionId =
masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1,
+ Lists.newArrayList(CatalogTestUtil.testTableId1),
"resource_group_quorum_failure",
+ transactionSource,
+ LoadJobSourceType.FRONTEND,
Config.stream_load_default_timeout_second);
+ List<TabletCommitInfo> commitInfos =
GlobalTransactionMgrTest.generateTabletCommitInfos(
+ CatalogTestUtil.testTabletId1,
+ Lists.newArrayList(CatalogTestUtil.testBackendId2,
CatalogTestUtil.testBackendId3));
+ try {
+
masterTransMgr.commitTransactionWithoutLock(CatalogTestUtil.testDbId1,
Lists.newArrayList(table),
+ transactionId, commitInfos, null);
+ Assert.fail();
+ } catch (TabletQuorumFailedException e) {
+ Assert.assertTrue(e.getMessage().contains("resource group
success quorum failed for group1"));
+ }
+
+ transactionId =
masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1,
+ Lists.newArrayList(CatalogTestUtil.testTableId1),
"resource_group_quorum_group2_failure",
+ transactionSource,
+ LoadJobSourceType.FRONTEND,
Config.stream_load_default_timeout_second);
+ commitInfos =
GlobalTransactionMgrTest.generateTabletCommitInfos(CatalogTestUtil.testTabletId1,
+ Lists.newArrayList(CatalogTestUtil.testBackendId1,
CatalogTestUtil.testBackendId2));
+ try {
+
masterTransMgr.commitTransactionWithoutLock(CatalogTestUtil.testDbId1,
Lists.newArrayList(table),
+ transactionId, commitInfos, null);
+ Assert.fail();
+ } catch (TabletQuorumFailedException e) {
+ Assert.assertTrue(e.getMessage().contains("resource group
success quorum failed for group2"));
+ }
+
+ backend2.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "group2"));
+
table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1,
+ new ReplicaAllocation(ImmutableMap.of(
+ Tag.createNotCheck(Tag.TYPE_LOCATION, "group1"),
(short) 1,
+ Tag.createNotCheck(Tag.TYPE_LOCATION, "group2"),
(short) 2)));
+ transactionId =
masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1,
+ Lists.newArrayList(CatalogTestUtil.testTableId1),
"resource_group_quorum_clamp", transactionSource,
+ LoadJobSourceType.FRONTEND,
Config.stream_load_default_timeout_second);
+ commitInfos =
GlobalTransactionMgrTest.generateTabletCommitInfos(CatalogTestUtil.testTabletId1,
+ Lists.newArrayList(CatalogTestUtil.testBackendId1,
CatalogTestUtil.testBackendId2));
+
masterTransMgr.commitTransactionWithoutLock(CatalogTestUtil.testDbId1,
Lists.newArrayList(table),
+ transactionId, commitInfos, null);
+
+ // Invalid items are ignored. The parse result is cached, so only
the first commit after
+ // a config change may warn; later commits on the hot path must
stay silent.
+ Config.resource_group_load_success_quorum = new String[]
{"invalid", "group1:not-a-number"};
+ try (TestLogAppender appender =
TestLogAppender.attach(DatabaseTransactionMgr.class, Level.WARN)) {
+ transactionId =
masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1,
+ Lists.newArrayList(CatalogTestUtil.testTableId1),
"resource_group_quorum_invalid_0",
+ transactionSource, LoadJobSourceType.FRONTEND,
Config.stream_load_default_timeout_second);
+
masterTransMgr.commitTransactionWithoutLock(CatalogTestUtil.testDbId1,
Lists.newArrayList(table),
+ transactionId, commitInfos, null);
+ Assert.assertTrue(appender.contains(Level.WARN, "Invalid
resource_group_load_success_quorum item"));
+ }
+ try (TestLogAppender appender =
TestLogAppender.attach(DatabaseTransactionMgr.class, Level.WARN)) {
+ transactionId =
masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1,
+ Lists.newArrayList(CatalogTestUtil.testTableId1),
"resource_group_quorum_invalid_1",
+ transactionSource, LoadJobSourceType.FRONTEND,
Config.stream_load_default_timeout_second);
+
masterTransMgr.commitTransactionWithoutLock(CatalogTestUtil.testDbId1,
Lists.newArrayList(table),
+ transactionId, commitInfos, null);
+ Assert.assertFalse(appender.contains(Level.WARN, "Invalid
resource_group_load_success_quorum item"));
+ }
+
+ } finally {
+ Config.resource_group_load_success_quorum =
originalResourceGroupSuccQuorum;
+
table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1,
originalAllocation);
+ backend1.setTagMap(backend1TagMap);
+ backend2.setTagMap(backend2TagMap);
+ backend3.setTagMap(backend3TagMap);
+ }
+ }
+
+ @Test
+ public void
testResourceGroupSuccessQuorumDoesNotShrinkAfterReplicaBecomesUnavailable()
throws UserException {
+ FakeEnv.setEnv(masterEnv);
+ String[] originalResourceGroupSuccQuorum =
Config.resource_group_load_success_quorum;
+ Backend backend1 =
masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId1);
+ Backend backend2 =
masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId2);
+ Backend backend3 =
masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId3);
+ Map<String, String> backend1TagMap =
ImmutableMap.copyOf(backend1.getTagMap());
+ Map<String, String> backend2TagMap =
ImmutableMap.copyOf(backend2.getTagMap());
+ Map<String, String> backend3TagMap =
ImmutableMap.copyOf(backend3.getTagMap());
+ backend1.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "group1"));
+ backend2.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "group1"));
+ backend3.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "group2"));
+ OlapTable table = (OlapTable) masterEnv.getInternalCatalog()
+ .getDbOrMetaException(CatalogTestUtil.testDbId1)
+ .getTableOrMetaException(CatalogTestUtil.testTableId1);
+ ReplicaAllocation originalAllocation = table.getPartitionInfo()
+ .getReplicaAllocation(CatalogTestUtil.testPartitionId1);
+
table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1,
+ new ReplicaAllocation(ImmutableMap.of(
+ Tag.createNotCheck(Tag.TYPE_LOCATION, "group1"),
(short) 2,
+ Tag.createNotCheck(Tag.TYPE_LOCATION, "group2"),
(short) 1)));
+ Replica backend2Replica =
table.getPartition(CatalogTestUtil.testPartitionId1)
+
.getIndex(CatalogTestUtil.testIndexId1).getTablet(CatalogTestUtil.testTabletId1)
+ .getReplicaByBackendId(CatalogTestUtil.testBackendId2);
+
+ try {
+ Config.resource_group_load_success_quorum = new String[]
{"group1:2"};
+ List<TabletCommitInfo> commitInfos =
GlobalTransactionMgrTest.generateTabletCommitInfos(
+ CatalogTestUtil.testTabletId1,
+ Lists.newArrayList(CatalogTestUtil.testBackendId1,
CatalogTestUtil.testBackendId3));
+ long transactionId =
masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1,
+ Lists.newArrayList(CatalogTestUtil.testTableId1),
"resource_group_quorum_dead_after_begin",
+ transactionSource, LoadJobSourceType.FRONTEND,
Config.stream_load_default_timeout_second);
+ backend2.setAlive(false);
+ try {
+
masterTransMgr.commitTransactionWithoutLock(CatalogTestUtil.testDbId1,
Lists.newArrayList(table),
+ transactionId, commitInfos, null);
+ Assert.fail();
+ } catch (TabletQuorumFailedException e) {
+ Assert.assertTrue(e.getMessage().contains("resource group
success quorum failed for group1"));
+ }
+ backend2.setAlive(true);
+
+ transactionId =
masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1,
+ Lists.newArrayList(CatalogTestUtil.testTableId1),
"resource_group_quorum_bad_after_begin",
+ transactionSource, LoadJobSourceType.FRONTEND,
Config.stream_load_default_timeout_second);
+ backend2Replica.setBad(true);
+ try {
+
masterTransMgr.commitTransactionWithoutLock(CatalogTestUtil.testDbId1,
Lists.newArrayList(table),
+ transactionId, commitInfos, null);
+ Assert.fail();
+ } catch (TabletQuorumFailedException e) {
+ Assert.assertTrue(e.getMessage().contains("resource group
success quorum failed for group1"));
+ }
+ } finally {
+ Config.resource_group_load_success_quorum =
originalResourceGroupSuccQuorum;
+ backend2.setAlive(true);
+ backend2Replica.setBad(false);
+
table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1,
originalAllocation);
+ backend1.setTagMap(backend1TagMap);
+ backend2.setTagMap(backend2TagMap);
+ backend3.setTagMap(backend3TagMap);
+ }
+ }
+
+ @Test
+ public void
testResourceGroupSuccessQuorumIgnoresExtraReplicaBeyondAllocation() throws
UserException {
+ FakeEnv.setEnv(masterEnv);
+ String[] originalResourceGroupSuccQuorum =
Config.resource_group_load_success_quorum;
+ Backend backend1 =
masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId1);
+ Backend backend2 =
masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId2);
+ Backend backend3 =
masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId3);
+ Map<String, String> backend1TagMap =
ImmutableMap.copyOf(backend1.getTagMap());
+ Map<String, String> backend2TagMap =
ImmutableMap.copyOf(backend2.getTagMap());
+ Map<String, String> backend3TagMap =
ImmutableMap.copyOf(backend3.getTagMap());
+ backend1.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "group1"));
+ backend2.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "group2"));
+ backend3.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "group2"));
+
+ long extraBackendId = CatalogTestUtil.testBackendId3 + 100;
+ Backend extraBackend = CatalogTestUtil.createBackend(extraBackendId,
"extra-host", 123, 124, 125);
+ extraBackend.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "group1"));
+ masterEnv.getCurrentSystemInfo().addBackend(extraBackend);
+
+ OlapTable table = (OlapTable) masterEnv.getInternalCatalog()
+ .getDbOrMetaException(CatalogTestUtil.testDbId1)
+ .getTableOrMetaException(CatalogTestUtil.testTableId1);
+ ReplicaAllocation originalAllocation = table.getPartitionInfo()
+ .getReplicaAllocation(CatalogTestUtil.testPartitionId1);
+
table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1,
+ new ReplicaAllocation(ImmutableMap.of(
+ Tag.createNotCheck(Tag.TYPE_LOCATION, "group1"),
(short) 1,
+ Tag.createNotCheck(Tag.TYPE_LOCATION, "group2"),
(short) 2)));
+ Tablet tablet = table.getPartition(CatalogTestUtil.testPartitionId1)
+
.getIndex(CatalogTestUtil.testIndexId1).getTablet(CatalogTestUtil.testTabletId1);
+ Replica extraReplica = new LocalReplica(CatalogTestUtil.testReplicaId3
+ 100,
+ extraBackendId, Replica.ReplicaState.NORMAL,
+ CatalogTestUtil.testStartVersion,
CatalogTestUtil.testSchemaHash1);
+ tablet.addReplica(extraReplica);
+
+ try {
+ Assert.assertEquals(3, table.getPartitionInfo()
+
.getReplicaAllocation(CatalogTestUtil.testPartitionId1).getTotalReplicaNum());
+ Assert.assertEquals(4, tablet.getReplicas().size());
+ Assert.assertEquals(Replica.ReplicaState.NORMAL,
+ tablet.getReplicaByBackendId(extraBackendId).getState());
+
+ Config.resource_group_load_success_quorum = new String[]
{"group1:2"};
+ long transactionId =
masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1,
+ Lists.newArrayList(CatalogTestUtil.testTableId1),
"resource_group_quorum_ignore_extra_replica",
+ transactionSource, LoadJobSourceType.FRONTEND,
Config.stream_load_default_timeout_second);
+ List<TabletCommitInfo> commitInfos =
GlobalTransactionMgrTest.generateTabletCommitInfos(
+ CatalogTestUtil.testTabletId1,
+ Lists.newArrayList(CatalogTestUtil.testBackendId1,
CatalogTestUtil.testBackendId2));
+
+
masterTransMgr.commitTransactionWithoutLock(CatalogTestUtil.testDbId1,
Lists.newArrayList(table),
+ transactionId, commitInfos, null);
+ } finally {
+ tablet.deleteReplica(extraReplica);
+ masterEnv.getCurrentSystemInfo().dropBackend(extraBackendId);
+ Config.resource_group_load_success_quorum =
originalResourceGroupSuccQuorum;
+
table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1,
originalAllocation);
+ backend1.setTagMap(backend1TagMap);
+ backend2.setTagMap(backend2TagMap);
+ backend3.setTagMap(backend3TagMap);
+ }
+ }
+
@Test
public void testAbortTransaction() throws UserException {
DatabaseTransactionMgr masterDbTransMgr =
masterTransMgr.getDatabaseTransactionMgr(CatalogTestUtil.testDbId1);
diff --git
a/regression-test/suites/load_p0/cross_az_quorum/test_resource_group_load_success_quorum.groovy
b/regression-test/suites/load_p0/cross_az_quorum/test_resource_group_load_success_quorum.groovy
new file mode 100644
index 00000000000..2e20bb5df25
--- /dev/null
+++
b/regression-test/suites/load_p0/cross_az_quorum/test_resource_group_load_success_quorum.groovy
@@ -0,0 +1,119 @@
+// 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.
+
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.NodeType
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+// No coverage for a slow remote AZ here: BE ends its first close-wait stage
at the ordinary
+// load_required_replica_num, which knows nothing about
resource_group_load_success_quorum, so a
+// replica that is merely slow can still be dropped and fail the commit.
Waiting per resource group
+// is the BE-side follow-up, deliberately not done yet.
+suite('test_resource_group_load_success_quorum', 'docker') {
+ def options = new ClusterOptions()
+ // BEs learn about a workload group only on the next topic publish, 30s
apart by default.
+ options.feConfigs += ['disable_tablet_scheduler=true',
'publish_topic_info_interval_ms=1000']
+ options.enableDebugPoints()
+ options.cloudMode = false
+
+ docker(options) {
+ def backends = sql_return_maparray('SHOW BACKENDS')
+ assertEquals(3, backends.size())
+ sql """ALTER SYSTEM MODIFY BACKEND '${backends[0].BackendId}' SET
('tag.location' = 'az1')"""
+ sql """ALTER SYSTEM MODIFY BACKEND '${backends[1].BackendId}' SET
('tag.location' = 'az1')"""
+ sql """ALTER SYSTEM MODIFY BACKEND '${backends[2].BackendId}' SET
('tag.location' = 'az2')"""
+
+ sql 'DROP TABLE IF EXISTS cross_az_quorum_table'
+ sql '''
+ CREATE TABLE cross_az_quorum_table (k INT)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES ('replication_allocation' = 'tag.location.az1: 2,
tag.location.az2: 1')
+ '''
+
+ // tag.location doubles as the compute group name: FE has to create
the `normal` workload
+ // group for a newly seen compute group and then publish it to the
BEs. Until both happen a
+ // load fails with "Can not find workload group normal in compute
group az1" (FE) or "not
+ // even find normal wg in BE". Probe with a real load -- SHOW WORKLOAD
GROUPS only proves
+ // the FE half. Turning workload groups off is not an option,
production runs with them on.
+ Awaitility.await().atMost(60, SECONDS).pollInterval(1, SECONDS).until({
+ try {
+ sql 'INSERT INTO cross_az_quorum_table VALUES (0)'
+ true
+ } catch (Exception e) {
+ false
+ }
+ })
+
+ def injectName = 'TxnManager.prepare_txn.random_failed'
+ GetDebugPoint().enableDebugPoint(backends[0].Host,
backends[0].HttpPort as int,
+ NodeType.BE, injectName, [percent: 1.0])
+
+ // The default empty config preserves the normal two-of-three quorum
behavior.
+ sql 'INSERT INTO cross_az_quorum_table VALUES (1)'
+
+ setFeConfig('resource_group_load_success_quorum', 'az1:2,az2:1')
+ test {
+ sql 'INSERT INTO cross_az_quorum_table VALUES (2)'
+ exception 'resource group success quorum failed for az1'
+ }
+
+ setFeConfig('resource_group_load_success_quorum', '')
+ sql 'INSERT INTO cross_az_quorum_table VALUES (3)'
+ GetDebugPoint().disableDebugPoint(backends[0].Host,
backends[0].HttpPort as int,
+ NodeType.BE, injectName)
+
+ // Losing all successful replicas in one AZ still leaves a normal
two-of-three quorum.
+ sql 'DROP TABLE IF EXISTS cross_az_quorum_az2_table'
+ sql '''
+ CREATE TABLE cross_az_quorum_az2_table (k INT)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES ('replication_allocation' = 'tag.location.az1: 2,
tag.location.az2: 1')
+ '''
+ GetDebugPoint().enableDebugPoint(backends[2].Host,
backends[2].HttpPort as int,
+ NodeType.BE, injectName, [percent: 1.0])
+ sql 'INSERT INTO cross_az_quorum_az2_table VALUES (1)'
+ setFeConfig('resource_group_load_success_quorum', 'az1:2,az2:1')
+ test {
+ sql 'INSERT INTO cross_az_quorum_az2_table VALUES (2)'
+ exception 'resource group success quorum failed for az2'
+ }
+ setFeConfig('resource_group_load_success_quorum', '')
+ GetDebugPoint().disableDebugPoint(backends[2].Host,
backends[2].HttpPort as int,
+ NodeType.BE, injectName)
+
+ sql 'DROP TABLE IF EXISTS cross_az_quorum_clamp_table'
+ sql '''
+ CREATE TABLE cross_az_quorum_clamp_table (k INT)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES ('replication_allocation' = 'tag.location.az1: 1')
+ '''
+ setFeConfig('resource_group_load_success_quorum', 'az1:2')
+ sql 'INSERT INTO cross_az_quorum_clamp_table VALUES (1)'
+
+ // A resource group the table does not place any replica in requires
nothing: the config is
+ // global, tables living in a single AZ must keep loading.
+ setFeConfig('resource_group_load_success_quorum', 'az1:1,az2:1')
+ sql 'INSERT INTO cross_az_quorum_clamp_table VALUES (2)'
+
+ // Invalid entries are ignored and must never break the commit path.
+ setFeConfig('resource_group_load_success_quorum',
'invalid,az1:not-a-number')
+ sql 'INSERT INTO cross_az_quorum_clamp_table VALUES (3)'
+ setFeConfig('resource_group_load_success_quorum', '')
+ }
+}
diff --git
a/regression-test/suites/load_p0/cross_az_quorum/test_resource_group_load_success_quorum_min_load_replica.groovy
b/regression-test/suites/load_p0/cross_az_quorum/test_resource_group_load_success_quorum_min_load_replica.groovy
new file mode 100644
index 00000000000..2514d9fefb3
--- /dev/null
+++
b/regression-test/suites/load_p0/cross_az_quorum/test_resource_group_load_success_quorum_min_load_replica.groovy
@@ -0,0 +1,122 @@
+// 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.
+
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.NodeType
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+// min_load_replica_num lowers how many successful replicas a commit needs,
which makes it easier
+// for the surviving replicas to all sit in one AZ.
resource_group_load_success_quorum is the guard for exactly
+// that combination: the two conditions stack, the AZ requirement never
relaxes the replica count.
+suite('test_resource_group_load_success_quorum_min_load_replica', 'docker') {
+ def options = new ClusterOptions()
+ // 5 backends so the table can hold 5 replicas: the default quorum is then
3 and lowering it
+ // to 2 actually changes the outcome. With 3 az1 and 2 az2 backends every
backend holds exactly
+ // one replica of the single tablet, which keeps the fault injection
deterministic.
+ options.beNum = 5
+ // BEs learn about a workload group only on the next topic publish, 30s
apart by default.
+ options.feConfigs += ['disable_tablet_scheduler=true',
'publish_topic_info_interval_ms=1000']
+ options.enableDebugPoints()
+ options.cloudMode = false
+
+ docker(options) {
+ def backends = sql_return_maparray('SHOW BACKENDS')
+ assertEquals(5, backends.size())
+ def az1Backends = backends[0..2]
+ def az2Backends = backends[3..4]
+ az1Backends.each {
+ sql """ALTER SYSTEM MODIFY BACKEND '${it.BackendId}' SET
('tag.location' = 'az1')"""
+ }
+ az2Backends.each {
+ sql """ALTER SYSTEM MODIFY BACKEND '${it.BackendId}' SET
('tag.location' = 'az2')"""
+ }
+
+ // tag.location doubles as the compute group name: FE has to create
the `normal` workload
+ // group for the new compute groups and then publish it to the BEs.
Probe with a real load,
+ // SHOW WORKLOAD GROUPS only proves the FE half.
+ sql 'DROP TABLE IF EXISTS cross_az_min_load_probe'
+ sql '''
+ CREATE TABLE cross_az_min_load_probe (k INT)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES ('replication_allocation' = 'tag.location.az1: 3,
tag.location.az2: 2')
+ '''
+ Awaitility.await().atMost(60, SECONDS).pollInterval(1, SECONDS).until({
+ try {
+ sql 'INSERT INTO cross_az_min_load_probe VALUES (0)'
+ true
+ } catch (Exception e) {
+ false
+ }
+ })
+
+ def injectName = 'TxnManager.prepare_txn.random_failed'
+ def enableInject = { List bes ->
+ bes.each {
+ GetDebugPoint().enableDebugPoint(it.Host, it.HttpPort as int,
NodeType.BE,
+ injectName, [percent: 1.0])
+ }
+ }
+ def disableInject = { List bes ->
+ bes.each {
+ GetDebugPoint().disableDebugPoint(it.Host, it.HttpPort as int,
NodeType.BE, injectName)
+ }
+ }
+
+ // Fail both az2 replicas and one az1 replica: 2 successful replicas
left, both in az1.
+ // Below the default quorum of 3, but enough for the lowered
min_load_replica_num of 2.
+ def singleAzFailures = az2Backends + [az1Backends[0]]
+ // Fail two az1 replicas and one az2 replica: 2 successful replicas
left, one per AZ.
+ def crossAzFailures = [az1Backends[0], az1Backends[1], az2Backends[0]]
+
+ def createTable = { String name ->
+ sql "DROP TABLE IF EXISTS ${name}"
+ sql """
+ CREATE TABLE ${name} (k INT)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES ('replication_allocation' = 'tag.location.az1: 3,
tag.location.az2: 2')
+ """
+ sql """ALTER TABLE ${name} SET ("min_load_replica_num" = "2")"""
+ }
+
+ // Without resource_group_load_success_quorum the lowered quorum
accepts a commit whose successful
+ // replicas all live in az1 -- the silent durability risk this feature
exists to remove.
+ enableInject(singleAzFailures)
+ createTable('cross_az_min_load_baseline')
+ sql 'INSERT INTO cross_az_min_load_baseline VALUES (1)'
+
+ // Same load, same lowered quorum, but now the AZ coverage requirement
rejects it.
+ createTable('cross_az_min_load_guarded')
+ setFeConfig('resource_group_load_success_quorum', 'az1:1,az2:1')
+ test {
+ sql 'INSERT INTO cross_az_min_load_guarded VALUES (1)'
+ exception 'resource group success quorum failed for az2'
+ }
+ disableInject(singleAzFailures)
+
+ // Still only 2 successful replicas and the same lowered quorum, but
this time they are
+ // spread over both AZs, so the commit is accepted: the AZ requirement
constrains where the
+ // successful replicas sit, it does not simply reject every degraded
load.
+ enableInject(crossAzFailures)
+ createTable('cross_az_min_load_spread')
+ sql 'INSERT INTO cross_az_min_load_spread VALUES (1)'
+ disableInject(crossAzFailures)
+
+ setFeConfig('resource_group_load_success_quorum', '')
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
