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..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 @@ -20,7 +20,6 @@ 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"; @@ -541,6 +540,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 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/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java index cb18868d0b78ed..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 @@ -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; @@ -120,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; @@ -494,6 +500,8 @@ private void checkCommitStatus(List tableList, TransactionState transacti TabletInvertedIndex tabletInvertedIndex = env.getTabletInvertedIndex(); Map> tabletToBackends = new HashMap<>(); Map idToTable = new HashMap<>(); + Map resourceGroupSuccQuorum = getResourceGroupSuccQuorum(); + Map backendLocationTags = resourceGroupSuccQuorum.isEmpty() ? 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 @@ private void checkCommitStatus(List
tableList, TransactionState transacti // (TODO): ignore the alter index if txn id is less than sc sched watermark int loadRequiredReplicaNum = table.getLoadRequiredReplicaNum(partition.getId()); + ReplicaAllocation replicaAllocation = resourceGroupSuccQuorum.isEmpty() ? null + : table.getPartitionInfo().getReplicaAllocation(partition.getId()); for (MaterializedIndex index : allIndices) { for (Tablet tablet : index.getTablets()) { tabletSuccReplicas.clear(); @@ -627,6 +637,12 @@ private void checkCommitStatus(List
tableList, TransactionState transacti throw new TransactionCommitFailedException("could not find replica for tablet [" + tabletId + "], backend [" + tabletBackend + "]"); } + if (!resourceGroupSuccQuorum.isEmpty()) { + backendLocationTags.computeIfAbsent(tabletBackend, backendId -> { + Backend backend = env.getCurrentSystemInfo().getBackend(backendId); + return backend == null ? "" : backend.getLocationTag().value; + }); + } // if the tablet have no replica's to commit or the tablet is a rolling up tablet, // the commit backends maybe null @@ -670,9 +686,71 @@ private void checkCommitStatus(List
tableList, TransactionState transacti throw new TabletQuorumFailedException(transactionId, errMsg); } + + for (Entry entry : resourceGroupSuccQuorum.entrySet()) { + String resourceGroup = entry.getKey(); + int replicaNumInResourceGroup = replicaAllocation.getReplicaNumByTag( + Tag.createNotCheck(Tag.TYPE_LOCATION, resourceGroup)); + int requiredInResourceGroup = Math.min(entry.getValue(), replicaNumInResourceGroup); + if (requiredInResourceGroup == 0) { + continue; + } + + int succInResourceGroup = 0; + for (Replica replica : tabletSuccReplicas) { + if (resourceGroup.equals( + backendLocationTags.get(replica.getBackendIdWithoutException()))) { + succInResourceGroup++; + } + } + if (succInResourceGroup < requiredInResourceGroup) { + String writeDetail = getTabletWriteDetail(tabletSuccReplicas, + tabletWriteFailedReplicas, tabletVersionFailedReplicas); + String errMsg = String.format("Failed to commit txn %s, cause tablet %s resource " + + "group success quorum failed for %s: required %s successful " + + "replicas, but only %s succeeded. table %s, partition: [ id=%s, " + + "commit version %s, visible version %s ], this tablet detail: %s. " + + "Please try again later.", transactionId, tablet.getId(), + resourceGroup, requiredInResourceGroup, succInResourceGroup, tableId, + partition.getId(), partition.getCommittedVersion(), + partition.getVisibleVersion(), writeDetail); + LOG.info(errMsg); + throw new TabletQuorumFailedException(transactionId, errMsg); + } + } + } + } + } + } + } + + private static Map getResourceGroupSuccQuorum() { + String[] config = Config.resource_group_load_success_quorum; + if (config == cachedResourceGroupSuccQuorumConfig) { + return cachedResourceGroupSuccQuorum; + } + synchronized (DatabaseTransactionMgr.class) { + config = Config.resource_group_load_success_quorum; + if (config == cachedResourceGroupSuccQuorumConfig) { + return cachedResourceGroupSuccQuorum; + } + Map parsedConfig = new HashMap<>(); + for (String item : config) { + String[] parts = item.split(":", -1); + try { + int configuredMin = Integer.parseInt(parts.length == 2 ? parts[1].trim() : ""); + if (parts[0].trim().isEmpty() || configuredMin < 0) { + throw new NumberFormatException(); } + parsedConfig.put(parts[0].trim(), configuredMin); + } catch (NumberFormatException e) { + LOG.warn("Invalid resource_group_load_success_quorum item '{}', ignored. Expected format " + + "resource_group:min_success_replicas with a non-negative integer.", item); } } + cachedResourceGroupSuccQuorum = parsedConfig; + cachedResourceGroupSuccQuorumConfig = config; + return parsedConfig; } } 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(); 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..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 @@ -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,230 @@ public void testNormal() throws UserException { 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 backend1TagMap = ImmutableMap.copyOf(backend1.getTagMap()); + Map backend2TagMap = ImmutableMap.copyOf(backend2.getTagMap()); + Map 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 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 backend1TagMap = ImmutableMap.copyOf(backend1.getTagMap()); + Map backend2TagMap = ImmutableMap.copyOf(backend2.getTagMap()); + Map 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 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 backend1TagMap = ImmutableMap.copyOf(backend1.getTagMap()); + Map backend2TagMap = ImmutableMap.copyOf(backend2.getTagMap()); + Map 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 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 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', '') + } +}