From b77e177174d2fde957c5d3a44f984159725788d1 Mon Sep 17 00:00:00 2001 From: deardeng Date: Fri, 14 Aug 2026 10:42:21 +0800 Subject: [PATCH 1/4] [feature](fe) Add cross-AZ success quorum check ### What problem does this PR solve? Issue Number: None 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 zone. Add a mutable FE configuration and enforce the per-AZ success floor in the centralized transaction commit check. Clamp each AZ 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 cross_az_succ_quorum to require a minimum number of successful load replicas per availability zone. ### Check List (For Author) - Test: Unit Test - DatabaseTransactionMgrTest cross-AZ quorum, unavailable replica, and extra replica cases - Behavior changed: Yes (when cross_az_succ_quorum is configured, FE rejects commits that do not meet the per-AZ success floor) - Does this need documentation: No --- .../java/org/apache/doris/common/Config.java | 43 ++++ .../transaction/DatabaseTransactionMgr.java | 44 ++++ .../DatabaseTransactionMgrTest.java | 229 ++++++++++++++++++ 3 files changed, 316 insertions(+) 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 3ce60affd466b9..6e2ed34afe99ed 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 @@ -17,9 +17,18 @@ package org.apache.doris.common; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + import java.io.File; +import java.util.HashMap; +import java.util.Map; public class Config extends ConfigBase { + private static final Logger LOG = LogManager.getLogger(Config.class); + // ConfigBase replaces the array on every update, so its identity is the cache version. + private static volatile String[] cachedCrossAzSuccQuorumConfig; + private static volatile Map cachedCrossAzSuccQuorum = Map.of(); @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") @@ -541,6 +550,40 @@ 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 availability zone for a load job.") + public static volatile String[] cross_az_succ_quorum = {}; + + public static Map getCrossAzSuccQuorum() { + String[] config = cross_az_succ_quorum; + if (config == cachedCrossAzSuccQuorumConfig) { + return cachedCrossAzSuccQuorum; + } + synchronized (Config.class) { + config = cross_az_succ_quorum; + if (config == cachedCrossAzSuccQuorumConfig) { + return cachedCrossAzSuccQuorum; + } + 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 cross_az_succ_quorum item '{}', ignored. Expected format " + + "az:min_success_replicas with a non-negative integer.", item); + } + } + cachedCrossAzSuccQuorum = parsedConfig; + cachedCrossAzSuccQuorumConfig = config; + return parsedConfig; + } + } + @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/transaction/DatabaseTransactionMgr.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java index cb18868d0b78ed..ce9b3a54584157 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.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.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; @@ -494,6 +497,8 @@ private void checkCommitStatus(List tableList, TransactionState transacti TabletInvertedIndex tabletInvertedIndex = env.getTabletInvertedIndex(); Map> tabletToBackends = new HashMap<>(); Map idToTable = new HashMap<>(); + Map crossAzSuccQuorum = Config.getCrossAzSuccQuorum(); + Map backendLocationTags = crossAzSuccQuorum.isEmpty() ? Map.of() : new HashMap<>(); for (int i = 0; i < tableList.size(); i++) { idToTable.put(tableList.get(i).getId(), tableList.get(i)); } @@ -610,6 +615,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 = crossAzSuccQuorum.isEmpty() ? null + : table.getPartitionInfo().getReplicaAllocation(partition.getId()); for (MaterializedIndex index : allIndices) { for (Tablet tablet : index.getTablets()) { tabletSuccReplicas.clear(); @@ -627,6 +634,12 @@ private void checkCommitStatus(List
tableList, TransactionState transacti throw new TransactionCommitFailedException("could not find replica for tablet [" + tabletId + "], backend [" + tabletBackend + "]"); } + if (!crossAzSuccQuorum.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,6 +683,37 @@ private void checkCommitStatus(List
tableList, TransactionState transacti throw new TabletQuorumFailedException(transactionId, errMsg); } + + for (Entry entry : crossAzSuccQuorum.entrySet()) { + String az = entry.getKey(); + int replicaNumInAz = replicaAllocation.getReplicaNumByTag( + Tag.createNotCheck(Tag.TYPE_LOCATION, az)); + int requiredInAz = Math.min(entry.getValue(), replicaNumInAz); + if (requiredInAz == 0) { + continue; + } + + int succInAz = 0; + for (Replica replica : tabletSuccReplicas) { + if (az.equals(backendLocationTags.get(replica.getBackendIdWithoutException()))) { + succInAz++; + } + } + if (succInAz < requiredInAz) { + String writeDetail = getTabletWriteDetail(tabletSuccReplicas, + tabletWriteFailedReplicas, tabletVersionFailedReplicas); + String errMsg = String.format("Failed to commit txn %s, cause tablet %s cross AZ " + + "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(), az, requiredInAz, succInAz, tableId, + partition.getId(), partition.getCommittedVersion(), + partition.getVisibleVersion(), writeDetail); + LOG.info(errMsg); + throw new TabletQuorumFailedException(transactionId, errMsg); + } + } } } } 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 7de6aa28aeac74..08453ad0e5d560 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.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.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,226 @@ public void testNormal() throws UserException { Assert.assertEquals(TransactionStatus.PREPARE, transactionState2.getTransactionStatus()); } + @Test + public void testCrossAzSuccessQuorum() throws UserException { + FakeEnv.setEnv(masterEnv); + String[] originalCrossAzSuccQuorum = Config.cross_az_succ_quorum; + Backend backend1 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId1); + Backend backend2 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId2); + Backend backend3 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId3); + Map backend1TagMap = ImmutableMap.copyOf(backend1.getTagMap()); + Map backend2TagMap = ImmutableMap.copyOf(backend2.getTagMap()); + Map backend3TagMap = ImmutableMap.copyOf(backend3.getTagMap()); + backend1.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az1")); + backend2.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az1")); + backend3.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az2")); + 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, "az1"), (short) 2, + Tag.createNotCheck(Tag.TYPE_LOCATION, "az2"), (short) 1))); + + try { + Config.cross_az_succ_quorum = new String[] {"az1:2", "az2:1"}; + Assert.assertEquals(ImmutableMap.of("az1", 2, "az2", 1), Config.getCrossAzSuccQuorum()); + long transactionId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1, + Lists.newArrayList(CatalogTestUtil.testTableId1), "cross_az_quorum_failure", transactionSource, + LoadJobSourceType.FRONTEND, Config.stream_load_default_timeout_second); + List 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("cross AZ success quorum failed for az1")); + } + + transactionId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1, + Lists.newArrayList(CatalogTestUtil.testTableId1), "cross_az_quorum_az2_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("cross AZ success quorum failed for az2")); + } + + backend2.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az2")); + table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1, + new ReplicaAllocation(ImmutableMap.of( + Tag.createNotCheck(Tag.TYPE_LOCATION, "az1"), (short) 1, + Tag.createNotCheck(Tag.TYPE_LOCATION, "az2"), (short) 2))); + transactionId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1, + Lists.newArrayList(CatalogTestUtil.testTableId1), "cross_az_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.cross_az_succ_quorum = new String[] {"invalid", "az1:not-a-number"}; + transactionId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1, + Lists.newArrayList(CatalogTestUtil.testTableId1), "cross_az_quorum_invalid_0", + transactionSource, LoadJobSourceType.FRONTEND, Config.stream_load_default_timeout_second); + masterTransMgr.commitTransactionWithoutLock(CatalogTestUtil.testDbId1, Lists.newArrayList(table), + transactionId, commitInfos, null); + try (TestLogAppender appender = TestLogAppender.attach(Config.class, Level.WARN)) { + transactionId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1, + Lists.newArrayList(CatalogTestUtil.testTableId1), "cross_az_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 cross_az_succ_quorum item")); + } + + } finally { + Config.cross_az_succ_quorum = originalCrossAzSuccQuorum; + table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1, originalAllocation); + backend1.setTagMap(backend1TagMap); + backend2.setTagMap(backend2TagMap); + backend3.setTagMap(backend3TagMap); + } + } + + @Test + public void testCrossAzSuccessQuorumDoesNotShrinkAfterReplicaBecomesUnavailable() throws UserException { + FakeEnv.setEnv(masterEnv); + String[] originalCrossAzSuccQuorum = Config.cross_az_succ_quorum; + Backend backend1 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId1); + Backend backend2 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId2); + Backend backend3 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId3); + Map backend1TagMap = ImmutableMap.copyOf(backend1.getTagMap()); + Map backend2TagMap = ImmutableMap.copyOf(backend2.getTagMap()); + Map backend3TagMap = ImmutableMap.copyOf(backend3.getTagMap()); + backend1.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az1")); + backend2.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az1")); + backend3.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az2")); + 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, "az1"), (short) 2, + Tag.createNotCheck(Tag.TYPE_LOCATION, "az2"), (short) 1))); + Replica backend2Replica = table.getPartition(CatalogTestUtil.testPartitionId1) + .getIndex(CatalogTestUtil.testIndexId1).getTablet(CatalogTestUtil.testTabletId1) + .getReplicaByBackendId(CatalogTestUtil.testBackendId2); + + try { + Config.cross_az_succ_quorum = new String[] {"az1:2"}; + List commitInfos = GlobalTransactionMgrTest.generateTabletCommitInfos( + CatalogTestUtil.testTabletId1, + Lists.newArrayList(CatalogTestUtil.testBackendId1, CatalogTestUtil.testBackendId3)); + long transactionId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1, + Lists.newArrayList(CatalogTestUtil.testTableId1), "cross_az_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("cross AZ success quorum failed for az1")); + } + backend2.setAlive(true); + + transactionId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1, + Lists.newArrayList(CatalogTestUtil.testTableId1), "cross_az_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("cross AZ success quorum failed for az1")); + } + } finally { + Config.cross_az_succ_quorum = originalCrossAzSuccQuorum; + backend2.setAlive(true); + backend2Replica.setBad(false); + table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1, originalAllocation); + backend1.setTagMap(backend1TagMap); + backend2.setTagMap(backend2TagMap); + backend3.setTagMap(backend3TagMap); + } + } + + @Test + public void testCrossAzSuccessQuorumIgnoresExtraReplicaBeyondAllocation() throws UserException { + FakeEnv.setEnv(masterEnv); + String[] originalCrossAzSuccQuorum = Config.cross_az_succ_quorum; + Backend backend1 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId1); + Backend backend2 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId2); + Backend backend3 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId3); + Map backend1TagMap = ImmutableMap.copyOf(backend1.getTagMap()); + Map backend2TagMap = ImmutableMap.copyOf(backend2.getTagMap()); + Map backend3TagMap = ImmutableMap.copyOf(backend3.getTagMap()); + backend1.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az1")); + backend2.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az2")); + backend3.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az2")); + + long extraBackendId = CatalogTestUtil.testBackendId3 + 100; + Backend extraBackend = CatalogTestUtil.createBackend(extraBackendId, "extra-host", 123, 124, 125); + extraBackend.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az1")); + 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, "az1"), (short) 1, + Tag.createNotCheck(Tag.TYPE_LOCATION, "az2"), (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.cross_az_succ_quorum = new String[] {"az1:2"}; + long transactionId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1, + Lists.newArrayList(CatalogTestUtil.testTableId1), "cross_az_quorum_ignore_extra_replica", + transactionSource, LoadJobSourceType.FRONTEND, Config.stream_load_default_timeout_second); + List 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.cross_az_succ_quorum = originalCrossAzSuccQuorum; + 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); From 270068ffbabaa6bc55eda192e08b6a83fb7fce94 Mon Sep 17 00:00:00 2001 From: deardeng Date: Fri, 14 Aug 2026 11:49:23 +0800 Subject: [PATCH 2/4] [fix](fe) Safely publish backend location updates ### What problem does this PR solve? Issue Number: None Related PR: #66751 Problem Summary: Backend location tags can be updated by MODIFY BACKEND while transaction commit threads read them. The plain locationTag reference had no Java memory-model publication edge, so readers could observe a stale availability-zone tag. Publish replacement Tag instances through a volatile reference and add a deterministic regression test for the required publication semantics. ### Release note None ### Check List (For Author) - Test: Unit Test - BackendTest#testLocationTagIsSafelyPublished - Behavior changed: No (ensures existing backend location updates are visible across threads) - Does this need documentation: No --- .../src/main/java/org/apache/doris/system/Backend.java | 3 +-- .../src/test/java/org/apache/doris/catalog/BackendTest.java | 6 ++++++ 2 files changed, 7 insertions(+), 2 deletions(-) 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 d403c88732ea2d..a4fe6e6105d25a 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/test/java/org/apache/doris/catalog/BackendTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/BackendTest.java index 1c5142b76254ce..a2f4b769ebd0a4 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 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 void getMethodTest() { Assert.assertTrue(backend.isAlive()); } + @Test + public void testLocationTagIsSafelyPublished() throws NoSuchFieldException { + Assert.assertTrue(Modifier.isVolatile(Backend.class.getDeclaredField("locationTag").getModifiers())); + } + @Test public void diskInfoTest() { Map diskInfos = new HashMap(); From dddbfd20196b88c771a107be94cd145fa0a54b76 Mon Sep 17 00:00:00 2001 From: deardeng Date: Mon, 17 Aug 2026 16:02:53 +0800 Subject: [PATCH 3/4] [refactor](fe) Refine resource group success quorum configuration ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: Replace availability-zone terminology with the existing resource group concept for load success quorum configuration. Keep Config focused on the raw configuration declaration and move transaction-specific parsing, caching, and logging into DatabaseTransactionMgr without changing runtime behavior. ### Release note None ### Check List (For Author) - Test: Unit Test - ./run-fe-ut.sh --run org.apache.doris.transaction.DatabaseTransactionMgrTest (19 tests passed) - mvn checkstyle:check -pl fe-common,fe-core (0 violations) - Behavior changed: No - Does this need documentation: No --- .../java/org/apache/doris/common/Config.java | 44 +------- .../transaction/DatabaseTransactionMgr.java | 74 ++++++++---- .../DatabaseTransactionMgrTest.java | 106 +++++++++--------- 3 files changed, 111 insertions(+), 113 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 6e2ed34afe99ed..b10ab57376f179 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 @@ -17,19 +17,9 @@ package org.apache.doris.common; -import org.apache.logging.log4j.LogManager; -import org.apache.logging.log4j.Logger; - import java.io.File; -import java.util.HashMap; -import java.util.Map; public class Config extends ConfigBase { - private static final Logger LOG = LogManager.getLogger(Config.class); - // ConfigBase replaces the array on every update, so its identity is the cache version. - private static volatile String[] cachedCrossAzSuccQuorumConfig; - private static volatile Map cachedCrossAzSuccQuorum = Map.of(); - @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"; @@ -551,38 +541,8 @@ public class Config extends ConfigBase { public static short min_load_replica_num = -1; @ConfField(mutable = true, masterOnly = true, description = "Minimum number of successfully written replicas " - + "required in each availability zone for a load job.") - public static volatile String[] cross_az_succ_quorum = {}; - - public static Map getCrossAzSuccQuorum() { - String[] config = cross_az_succ_quorum; - if (config == cachedCrossAzSuccQuorumConfig) { - return cachedCrossAzSuccQuorum; - } - synchronized (Config.class) { - config = cross_az_succ_quorum; - if (config == cachedCrossAzSuccQuorumConfig) { - return cachedCrossAzSuccQuorum; - } - 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 cross_az_succ_quorum item '{}', ignored. Expected format " - + "az:min_success_replicas with a non-negative integer.", item); - } - } - cachedCrossAzSuccQuorum = parsedConfig; - cachedCrossAzSuccQuorumConfig = config; - return parsedConfig; - } - } + + "required in each resource group for a load job.") + public static volatile String[] resource_group_succ_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/transaction/DatabaseTransactionMgr.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java index ce9b3a54584157..1d15400feb2cb6 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 @@ -123,6 +123,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 = Map.of(); private final long dbId; @@ -497,8 +500,8 @@ private void checkCommitStatus(List
tableList, TransactionState transacti TabletInvertedIndex tabletInvertedIndex = env.getTabletInvertedIndex(); Map> tabletToBackends = new HashMap<>(); Map idToTable = new HashMap<>(); - Map crossAzSuccQuorum = Config.getCrossAzSuccQuorum(); - Map backendLocationTags = crossAzSuccQuorum.isEmpty() ? Map.of() : new HashMap<>(); + Map resourceGroupSuccQuorum = getResourceGroupSuccQuorum(); + Map backendLocationTags = resourceGroupSuccQuorum.isEmpty() ? Map.of() : new HashMap<>(); for (int i = 0; i < tableList.size(); i++) { idToTable.put(tableList.get(i).getId(), tableList.get(i)); } @@ -615,7 +618,7 @@ 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 = crossAzSuccQuorum.isEmpty() ? null + ReplicaAllocation replicaAllocation = resourceGroupSuccQuorum.isEmpty() ? null : table.getPartitionInfo().getReplicaAllocation(partition.getId()); for (MaterializedIndex index : allIndices) { for (Tablet tablet : index.getTablets()) { @@ -634,7 +637,7 @@ private void checkCommitStatus(List
tableList, TransactionState transacti throw new TransactionCommitFailedException("could not find replica for tablet [" + tabletId + "], backend [" + tabletBackend + "]"); } - if (!crossAzSuccQuorum.isEmpty()) { + if (!resourceGroupSuccQuorum.isEmpty()) { backendLocationTags.computeIfAbsent(tabletBackend, backendId -> { Backend backend = env.getCurrentSystemInfo().getBackend(backendId); return backend == null ? "" : backend.getLocationTag().value; @@ -684,30 +687,31 @@ private void checkCommitStatus(List
tableList, TransactionState transacti throw new TabletQuorumFailedException(transactionId, errMsg); } - for (Entry entry : crossAzSuccQuorum.entrySet()) { - String az = entry.getKey(); - int replicaNumInAz = replicaAllocation.getReplicaNumByTag( - Tag.createNotCheck(Tag.TYPE_LOCATION, az)); - int requiredInAz = Math.min(entry.getValue(), replicaNumInAz); - if (requiredInAz == 0) { + 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 succInAz = 0; + int succInResourceGroup = 0; for (Replica replica : tabletSuccReplicas) { - if (az.equals(backendLocationTags.get(replica.getBackendIdWithoutException()))) { - succInAz++; + if (resourceGroup.equals( + backendLocationTags.get(replica.getBackendIdWithoutException()))) { + succInResourceGroup++; } } - if (succInAz < requiredInAz) { + if (succInResourceGroup < requiredInResourceGroup) { String writeDetail = getTabletWriteDetail(tabletSuccReplicas, tabletWriteFailedReplicas, tabletVersionFailedReplicas); - String errMsg = String.format("Failed to commit txn %s, cause tablet %s cross AZ " - + "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(), az, requiredInAz, succInAz, tableId, + 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); @@ -720,6 +724,36 @@ private void checkCommitStatus(List
tableList, TransactionState transacti } } + private static Map getResourceGroupSuccQuorum() { + String[] config = Config.resource_group_succ_quorum; + if (config == cachedResourceGroupSuccQuorumConfig) { + return cachedResourceGroupSuccQuorum; + } + synchronized (DatabaseTransactionMgr.class) { + config = Config.resource_group_succ_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_succ_quorum item '{}', ignored. Expected format " + + "resource_group:min_success_replicas with a non-negative integer.", item); + } + } + cachedResourceGroupSuccQuorum = parsedConfig; + cachedResourceGroupSuccQuorumConfig = config; + return parsedConfig; + } + } + private String getTabletWriteDetail(List tabletSuccReplicas, List tabletWriteFailedReplicas, List tabletVersionFailedReplicas) { String writeDetail = ""; 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 08453ad0e5d560..c398fb73caa281 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 @@ -244,32 +244,32 @@ public void testNormal() throws UserException { } @Test - public void testCrossAzSuccessQuorum() throws UserException { + public void testResourceGroupSuccessQuorum() throws UserException { FakeEnv.setEnv(masterEnv); - String[] originalCrossAzSuccQuorum = Config.cross_az_succ_quorum; + String[] originalResourceGroupSuccQuorum = Config.resource_group_succ_quorum; Backend backend1 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId1); Backend backend2 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId2); Backend backend3 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId3); Map backend1TagMap = ImmutableMap.copyOf(backend1.getTagMap()); Map backend2TagMap = ImmutableMap.copyOf(backend2.getTagMap()); Map backend3TagMap = ImmutableMap.copyOf(backend3.getTagMap()); - backend1.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az1")); - backend2.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az1")); - backend3.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az2")); + 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, "az1"), (short) 2, - Tag.createNotCheck(Tag.TYPE_LOCATION, "az2"), (short) 1))); + Tag.createNotCheck(Tag.TYPE_LOCATION, "group1"), (short) 2, + Tag.createNotCheck(Tag.TYPE_LOCATION, "group2"), (short) 1))); try { - Config.cross_az_succ_quorum = new String[] {"az1:2", "az2:1"}; - Assert.assertEquals(ImmutableMap.of("az1", 2, "az2", 1), Config.getCrossAzSuccQuorum()); + Config.resource_group_succ_quorum = new String[] {"group1:2", "group2:1"}; long transactionId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1, - Lists.newArrayList(CatalogTestUtil.testTableId1), "cross_az_quorum_failure", transactionSource, + Lists.newArrayList(CatalogTestUtil.testTableId1), "resource_group_quorum_failure", + transactionSource, LoadJobSourceType.FRONTEND, Config.stream_load_default_timeout_second); List commitInfos = GlobalTransactionMgrTest.generateTabletCommitInfos( CatalogTestUtil.testTabletId1, @@ -279,11 +279,12 @@ public void testCrossAzSuccessQuorum() throws UserException { transactionId, commitInfos, null); Assert.fail(); } catch (TabletQuorumFailedException e) { - Assert.assertTrue(e.getMessage().contains("cross AZ success quorum failed for az1")); + Assert.assertTrue(e.getMessage().contains("resource group success quorum failed for group1")); } transactionId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1, - Lists.newArrayList(CatalogTestUtil.testTableId1), "cross_az_quorum_az2_failure", transactionSource, + 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)); @@ -292,16 +293,16 @@ public void testCrossAzSuccessQuorum() throws UserException { transactionId, commitInfos, null); Assert.fail(); } catch (TabletQuorumFailedException e) { - Assert.assertTrue(e.getMessage().contains("cross AZ success quorum failed for az2")); + Assert.assertTrue(e.getMessage().contains("resource group success quorum failed for group2")); } - backend2.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az2")); + backend2.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "group2")); table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1, new ReplicaAllocation(ImmutableMap.of( - Tag.createNotCheck(Tag.TYPE_LOCATION, "az1"), (short) 1, - Tag.createNotCheck(Tag.TYPE_LOCATION, "az2"), (short) 2))); + 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), "cross_az_quorum_clamp", transactionSource, + 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)); @@ -310,23 +311,26 @@ public void testCrossAzSuccessQuorum() throws UserException { // 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.cross_az_succ_quorum = new String[] {"invalid", "az1:not-a-number"}; - transactionId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1, - Lists.newArrayList(CatalogTestUtil.testTableId1), "cross_az_quorum_invalid_0", - transactionSource, LoadJobSourceType.FRONTEND, Config.stream_load_default_timeout_second); - masterTransMgr.commitTransactionWithoutLock(CatalogTestUtil.testDbId1, Lists.newArrayList(table), - transactionId, commitInfos, null); - try (TestLogAppender appender = TestLogAppender.attach(Config.class, Level.WARN)) { + Config.resource_group_succ_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_succ_quorum item")); + } + try (TestLogAppender appender = TestLogAppender.attach(DatabaseTransactionMgr.class, Level.WARN)) { transactionId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1, - Lists.newArrayList(CatalogTestUtil.testTableId1), "cross_az_quorum_invalid_1", + 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 cross_az_succ_quorum item")); + Assert.assertFalse(appender.contains(Level.WARN, "Invalid resource_group_succ_quorum item")); } } finally { - Config.cross_az_succ_quorum = originalCrossAzSuccQuorum; + Config.resource_group_succ_quorum = originalResourceGroupSuccQuorum; table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1, originalAllocation); backend1.setTagMap(backend1TagMap); backend2.setTagMap(backend2TagMap); @@ -335,18 +339,18 @@ public void testCrossAzSuccessQuorum() throws UserException { } @Test - public void testCrossAzSuccessQuorumDoesNotShrinkAfterReplicaBecomesUnavailable() throws UserException { + public void testResourceGroupSuccessQuorumDoesNotShrinkAfterReplicaBecomesUnavailable() throws UserException { FakeEnv.setEnv(masterEnv); - String[] originalCrossAzSuccQuorum = Config.cross_az_succ_quorum; + String[] originalResourceGroupSuccQuorum = Config.resource_group_succ_quorum; Backend backend1 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId1); Backend backend2 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId2); Backend backend3 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId3); Map backend1TagMap = ImmutableMap.copyOf(backend1.getTagMap()); Map backend2TagMap = ImmutableMap.copyOf(backend2.getTagMap()); Map backend3TagMap = ImmutableMap.copyOf(backend3.getTagMap()); - backend1.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az1")); - backend2.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az1")); - backend3.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az2")); + 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); @@ -354,19 +358,19 @@ public void testCrossAzSuccessQuorumDoesNotShrinkAfterReplicaBecomesUnavailable( .getReplicaAllocation(CatalogTestUtil.testPartitionId1); table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1, new ReplicaAllocation(ImmutableMap.of( - Tag.createNotCheck(Tag.TYPE_LOCATION, "az1"), (short) 2, - Tag.createNotCheck(Tag.TYPE_LOCATION, "az2"), (short) 1))); + 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.cross_az_succ_quorum = new String[] {"az1:2"}; + Config.resource_group_succ_quorum = new String[] {"group1:2"}; List commitInfos = GlobalTransactionMgrTest.generateTabletCommitInfos( CatalogTestUtil.testTabletId1, Lists.newArrayList(CatalogTestUtil.testBackendId1, CatalogTestUtil.testBackendId3)); long transactionId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1, - Lists.newArrayList(CatalogTestUtil.testTableId1), "cross_az_quorum_dead_after_begin", + Lists.newArrayList(CatalogTestUtil.testTableId1), "resource_group_quorum_dead_after_begin", transactionSource, LoadJobSourceType.FRONTEND, Config.stream_load_default_timeout_second); backend2.setAlive(false); try { @@ -374,12 +378,12 @@ public void testCrossAzSuccessQuorumDoesNotShrinkAfterReplicaBecomesUnavailable( transactionId, commitInfos, null); Assert.fail(); } catch (TabletQuorumFailedException e) { - Assert.assertTrue(e.getMessage().contains("cross AZ success quorum failed for az1")); + Assert.assertTrue(e.getMessage().contains("resource group success quorum failed for group1")); } backend2.setAlive(true); transactionId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1, - Lists.newArrayList(CatalogTestUtil.testTableId1), "cross_az_quorum_bad_after_begin", + Lists.newArrayList(CatalogTestUtil.testTableId1), "resource_group_quorum_bad_after_begin", transactionSource, LoadJobSourceType.FRONTEND, Config.stream_load_default_timeout_second); backend2Replica.setBad(true); try { @@ -387,10 +391,10 @@ public void testCrossAzSuccessQuorumDoesNotShrinkAfterReplicaBecomesUnavailable( transactionId, commitInfos, null); Assert.fail(); } catch (TabletQuorumFailedException e) { - Assert.assertTrue(e.getMessage().contains("cross AZ success quorum failed for az1")); + Assert.assertTrue(e.getMessage().contains("resource group success quorum failed for group1")); } } finally { - Config.cross_az_succ_quorum = originalCrossAzSuccQuorum; + Config.resource_group_succ_quorum = originalResourceGroupSuccQuorum; backend2.setAlive(true); backend2Replica.setBad(false); table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1, originalAllocation); @@ -401,22 +405,22 @@ public void testCrossAzSuccessQuorumDoesNotShrinkAfterReplicaBecomesUnavailable( } @Test - public void testCrossAzSuccessQuorumIgnoresExtraReplicaBeyondAllocation() throws UserException { + public void testResourceGroupSuccessQuorumIgnoresExtraReplicaBeyondAllocation() throws UserException { FakeEnv.setEnv(masterEnv); - String[] originalCrossAzSuccQuorum = Config.cross_az_succ_quorum; + String[] originalResourceGroupSuccQuorum = Config.resource_group_succ_quorum; Backend backend1 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId1); Backend backend2 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId2); Backend backend3 = masterEnv.getCurrentSystemInfo().getBackend(CatalogTestUtil.testBackendId3); Map backend1TagMap = ImmutableMap.copyOf(backend1.getTagMap()); Map backend2TagMap = ImmutableMap.copyOf(backend2.getTagMap()); Map backend3TagMap = ImmutableMap.copyOf(backend3.getTagMap()); - backend1.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az1")); - backend2.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az2")); - backend3.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "az2")); + 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, "az1")); + extraBackend.setTagMap(ImmutableMap.of(Tag.TYPE_LOCATION, "group1")); masterEnv.getCurrentSystemInfo().addBackend(extraBackend); OlapTable table = (OlapTable) masterEnv.getInternalCatalog() @@ -426,8 +430,8 @@ public void testCrossAzSuccessQuorumIgnoresExtraReplicaBeyondAllocation() throws .getReplicaAllocation(CatalogTestUtil.testPartitionId1); table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1, new ReplicaAllocation(ImmutableMap.of( - Tag.createNotCheck(Tag.TYPE_LOCATION, "az1"), (short) 1, - Tag.createNotCheck(Tag.TYPE_LOCATION, "az2"), (short) 2))); + 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, @@ -442,9 +446,9 @@ public void testCrossAzSuccessQuorumIgnoresExtraReplicaBeyondAllocation() throws Assert.assertEquals(Replica.ReplicaState.NORMAL, tablet.getReplicaByBackendId(extraBackendId).getState()); - Config.cross_az_succ_quorum = new String[] {"az1:2"}; + Config.resource_group_succ_quorum = new String[] {"group1:2"}; long transactionId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1, - Lists.newArrayList(CatalogTestUtil.testTableId1), "cross_az_quorum_ignore_extra_replica", + Lists.newArrayList(CatalogTestUtil.testTableId1), "resource_group_quorum_ignore_extra_replica", transactionSource, LoadJobSourceType.FRONTEND, Config.stream_load_default_timeout_second); List commitInfos = GlobalTransactionMgrTest.generateTabletCommitInfos( CatalogTestUtil.testTabletId1, @@ -455,7 +459,7 @@ public void testCrossAzSuccessQuorumIgnoresExtraReplicaBeyondAllocation() throws } finally { tablet.deleteReplica(extraReplica); masterEnv.getCurrentSystemInfo().dropBackend(extraBackendId); - Config.cross_az_succ_quorum = originalCrossAzSuccQuorum; + Config.resource_group_succ_quorum = originalResourceGroupSuccQuorum; table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1, originalAllocation); backend1.setTagMap(backend1TagMap); backend2.setTagMap(backend2TagMap); From 6a6e85d357e4fcfe66a7d033956887e26c0c6f4f Mon Sep 17 00:00:00 2001 From: deardeng Date: Tue, 18 Aug 2026 20:26:37 +0800 Subject: [PATCH 4/4] [improvement](fe) Clarify load success quorum configuration Issue Number: None Related PR: #66751, #66827 Problem Summary: The FE configuration name resource_group_succ_quorum did not identify that the requirement applies to successful load writes. Rename it to resource_group_load_success_quorum and update parsing, logging, and unit-test references. Migrate the two cross-resource-group load quorum regression cases from the BE companion change, adapting them to the renamed FE configuration and current FE error text. None - Test: No need to test (user requested no UT or regression execution); git diff --check passed. - Behavior changed: No (configuration key is renamed before release; regression coverage is migrated). - Does this need documentation: No --- .../java/org/apache/doris/common/Config.java | 2 +- .../transaction/DatabaseTransactionMgr.java | 6 +- .../DatabaseTransactionMgrTest.java | 24 ++-- ..._resource_group_load_success_quorum.groovy | 119 +++++++++++++++++ ...oad_success_quorum_min_load_replica.groovy | 122 ++++++++++++++++++ 5 files changed, 257 insertions(+), 16 deletions(-) create mode 100644 regression-test/suites/load_p0/cross_az_quorum/test_resource_group_load_success_quorum.groovy create mode 100644 regression-test/suites/load_p0/cross_az_quorum/test_resource_group_load_success_quorum_min_load_replica.groovy 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 b10ab57376f179..1a4eb7898d5a1a 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 @@ -542,7 +542,7 @@ public class Config extends ConfigBase { @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_succ_quorum = {}; + 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/transaction/DatabaseTransactionMgr.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java index 1d15400feb2cb6..9ad29be502fec1 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 @@ -725,12 +725,12 @@ private void checkCommitStatus(List
tableList, TransactionState transacti } private static Map getResourceGroupSuccQuorum() { - String[] config = Config.resource_group_succ_quorum; + String[] config = Config.resource_group_load_success_quorum; if (config == cachedResourceGroupSuccQuorumConfig) { return cachedResourceGroupSuccQuorum; } synchronized (DatabaseTransactionMgr.class) { - config = Config.resource_group_succ_quorum; + config = Config.resource_group_load_success_quorum; if (config == cachedResourceGroupSuccQuorumConfig) { return cachedResourceGroupSuccQuorum; } @@ -744,7 +744,7 @@ private static Map getResourceGroupSuccQuorum() { } parsedConfig.put(parts[0].trim(), configuredMin); } catch (NumberFormatException e) { - LOG.warn("Invalid resource_group_succ_quorum item '{}', ignored. Expected format " + LOG.warn("Invalid resource_group_load_success_quorum item '{}', ignored. Expected format " + "resource_group:min_success_replicas with a non-negative integer.", item); } } 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 c398fb73caa281..bc85c2134bf1fe 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 @@ -246,7 +246,7 @@ public void testNormal() throws UserException { @Test public void testResourceGroupSuccessQuorum() throws UserException { FakeEnv.setEnv(masterEnv); - String[] originalResourceGroupSuccQuorum = Config.resource_group_succ_quorum; + 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); @@ -266,7 +266,7 @@ public void testResourceGroupSuccessQuorum() throws UserException { Tag.createNotCheck(Tag.TYPE_LOCATION, "group2"), (short) 1))); try { - Config.resource_group_succ_quorum = new String[] {"group1:2", "group2:1"}; + 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, @@ -311,14 +311,14 @@ public void testResourceGroupSuccessQuorum() throws UserException { // 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_succ_quorum = new String[] {"invalid", "group1:not-a-number"}; + 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_succ_quorum item")); + 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, @@ -326,11 +326,11 @@ public void testResourceGroupSuccessQuorum() throws UserException { 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_succ_quorum item")); + Assert.assertFalse(appender.contains(Level.WARN, "Invalid resource_group_load_success_quorum item")); } } finally { - Config.resource_group_succ_quorum = originalResourceGroupSuccQuorum; + Config.resource_group_load_success_quorum = originalResourceGroupSuccQuorum; table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1, originalAllocation); backend1.setTagMap(backend1TagMap); backend2.setTagMap(backend2TagMap); @@ -341,7 +341,7 @@ public void testResourceGroupSuccessQuorum() throws UserException { @Test public void testResourceGroupSuccessQuorumDoesNotShrinkAfterReplicaBecomesUnavailable() throws UserException { FakeEnv.setEnv(masterEnv); - String[] originalResourceGroupSuccQuorum = Config.resource_group_succ_quorum; + 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); @@ -365,7 +365,7 @@ public void testResourceGroupSuccessQuorumDoesNotShrinkAfterReplicaBecomesUnavai .getReplicaByBackendId(CatalogTestUtil.testBackendId2); try { - Config.resource_group_succ_quorum = new String[] {"group1:2"}; + Config.resource_group_load_success_quorum = new String[] {"group1:2"}; List commitInfos = GlobalTransactionMgrTest.generateTabletCommitInfos( CatalogTestUtil.testTabletId1, Lists.newArrayList(CatalogTestUtil.testBackendId1, CatalogTestUtil.testBackendId3)); @@ -394,7 +394,7 @@ public void testResourceGroupSuccessQuorumDoesNotShrinkAfterReplicaBecomesUnavai Assert.assertTrue(e.getMessage().contains("resource group success quorum failed for group1")); } } finally { - Config.resource_group_succ_quorum = originalResourceGroupSuccQuorum; + Config.resource_group_load_success_quorum = originalResourceGroupSuccQuorum; backend2.setAlive(true); backend2Replica.setBad(false); table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1, originalAllocation); @@ -407,7 +407,7 @@ public void testResourceGroupSuccessQuorumDoesNotShrinkAfterReplicaBecomesUnavai @Test public void testResourceGroupSuccessQuorumIgnoresExtraReplicaBeyondAllocation() throws UserException { FakeEnv.setEnv(masterEnv); - String[] originalResourceGroupSuccQuorum = Config.resource_group_succ_quorum; + 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); @@ -446,7 +446,7 @@ public void testResourceGroupSuccessQuorumIgnoresExtraReplicaBeyondAllocation() Assert.assertEquals(Replica.ReplicaState.NORMAL, tablet.getReplicaByBackendId(extraBackendId).getState()); - Config.resource_group_succ_quorum = new String[] {"group1:2"}; + 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); @@ -459,7 +459,7 @@ public void testResourceGroupSuccessQuorumIgnoresExtraReplicaBeyondAllocation() } finally { tablet.deleteReplica(extraReplica); masterEnv.getCurrentSystemInfo().dropBackend(extraBackendId); - Config.resource_group_succ_quorum = originalResourceGroupSuccQuorum; + Config.resource_group_load_success_quorum = originalResourceGroupSuccQuorum; table.getPartitionInfo().setReplicaAllocation(CatalogTestUtil.testPartitionId1, originalAllocation); backend1.setTagMap(backend1TagMap); backend2.setTagMap(backend2TagMap); 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 00000000000000..2e20bb5df254af --- /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 00000000000000..2514d9fefb38a2 --- /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', '') + } +}