Skip to content

Commit 961cd32

Browse files
deardengdeardeng
committed
[feature](fe) Add resource group affinity extension framework
Issue Number: None Related PR: None Problem Summary: Add a public Resource Group affinity extension framework for FE query, load, and repair scheduling paths while keeping the default community implementation as no-op. The change introduces shared policy interfaces, no-op factories, session/config/thrift fields, FE local resource group propagation, and observability fields so downstream implementations can plug in real affinity behavior without changing Apache default scheduling semantics. The audit table schema and AuditLoader buffer now also include the new affinity audit fields so internal.__internal_schema.audit_log stays consistent with the text audit event. EXPLAIN INSERT and group commit load backend selection now use the same load-affinity enable gate as the real load scheduling paths, so disabled load affinity does not call downstream load decision hooks. Forwarded group commit requests only deserialize affinity decisions when the optional affinity fields are present. Scan range planning preserves the invalid-compute-group short-circuit before checking backend membership. Add a no-op Resource Group affinity extension framework and related compatibility fields. - Test: Unit Test / Regression test not run - mvn checkstyle:check -pl fe-core - mvn test -pl fe-core -am -Dtest=ResourceGroupAffinityNoOpPolicyTest,SessionVariableAffinityTest,OlapScanNodeAffinityConfigTest,TabletSchedCtxTest#testRepairSrcAffinityConfigGate -DfailIfNoTests=false - mvn -Dmaven.build.cache.enabled=false test -pl fe-core -am -Dtest=InternalSchemaInitializerTest,AuditLoaderTest -DfailIfNoTests=false - mvn -Dmaven.build.cache.enabled=false test -pl fe-core -am -Dtest=OlapTableSinkAffinityExplainTest -DfailIfNoTests=false - mvn -Dmaven.build.cache.enabled=false test -pl fe-core -am -Dtest=OlapScanNodeAffinityConfigTest,GroupCommitManagerAffinityTest,MasterOpExecutorAffinityTest,FrontendServiceImplAffinityTest,OlapTableSinkAffinityExplainTest -DfailIfNoTests=false - git diff --check - Regression test not run: only updated frontends() TVF assertion/projection for the new column and no local Doris cluster was started. - Behavior changed: No - Does this need documentation: No
1 parent de203c5 commit 961cd32

49 files changed

Lines changed: 2361 additions & 131 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎bin/start_fe.sh‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@ OPTS="$(getopt \
3131
-l 'daemon' \
3232
-l 'helper:' \
3333
-l 'image:' \
34+
-l 'local_resource_group:' \
3435
-l 'version' \
3536
-l 'metadata_failure_recovery' \
3637
-l 'recovery_journal_id:' \
@@ -47,6 +48,7 @@ HELPER=''
4748
IMAGE_PATH=''
4849
IMAGE_TOOL=''
4950
OPT_VERSION=''
51+
declare -a LOCAL_RESOURCE_GROUP_ARGS=()
5052
declare -a HELPER_ARGS=()
5153
declare -a METADATA_FAILURE_RECOVERY_ARGS=()
5254
declare -a RECOVERY_JOURNAL_ID_ARGS=()
@@ -83,6 +85,10 @@ while true; do
8385
IMAGE_PATH="$2"
8486
shift 2
8587
;;
88+
--local_resource_group)
89+
LOCAL_RESOURCE_GROUP_ARGS=("--local_resource_group" "$2")
90+
shift 2
91+
;;
8692
--cluster_snapshot)
8793
CLUSTER_SNAPSHOT_ARGS=("--cluster_snapshot" "$2")
8894
shift 2
@@ -428,23 +434,23 @@ fi
428434

429435
if [[ "${OPT_VERSION}" != "" ]]; then
430436
export DORIS_LOG_TO_STDERR=1
431-
${LIMIT:+${LIMIT}} "${JAVA}" org.apache.doris.DorisFE --version
437+
${LIMIT:+${LIMIT}} "${JAVA}" org.apache.doris.DorisFE "${LOCAL_RESOURCE_GROUP_ARGS[@]}" --version
432438
exit 0
433439
fi
434440

435441
if [[ "${IMAGE_TOOL}" -eq 1 ]]; then
436442
if [[ -n "${IMAGE_PATH}" ]]; then
437-
${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE -i "${IMAGE_PATH}"
443+
${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${LOCAL_RESOURCE_GROUP_ARGS[@]}" -i "${IMAGE_PATH}"
438444
else
439445
echo "Internal error, USE IMAGE_TOOL like: ./start_fe.sh --image image_path"
440446
fi
441447
elif [[ "${RUN_DAEMON}" -eq 1 ]]; then
442-
nohup ${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "${DROP_BACKENDS_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" 2>&1 </dev/null &
448+
nohup ${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" "${LOCAL_RESOURCE_GROUP_ARGS[@]}" "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "${DROP_BACKENDS_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" 2>&1 </dev/null &
443449
elif [[ "${RUN_CONSOLE}" -eq 1 ]]; then
444450
export DORIS_LOG_TO_STDERR=1
445-
${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" ${OPT_VERSION:+${OPT_VERSION}} "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "${DROP_BACKENDS_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" </dev/null
451+
${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" "${LOCAL_RESOURCE_GROUP_ARGS[@]}" ${OPT_VERSION:+${OPT_VERSION}} "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "${DROP_BACKENDS_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" </dev/null
446452
else
447-
${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" ${OPT_VERSION:+${OPT_VERSION}} "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "${DROP_BACKENDS_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" 2>&1 </dev/null
453+
${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" "${LOCAL_RESOURCE_GROUP_ARGS[@]}" ${OPT_VERSION:+${OPT_VERSION}} "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "${DROP_BACKENDS_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" 2>&1 </dev/null
448454
fi
449455

450456
if [[ "${OPT_VERSION}" != "" ]]; then

‎fe/fe-common/src/main/java/org/apache/doris/common/Config.java‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -135,6 +135,19 @@ public class Config extends ConfigBase {
135135
description = {"Whether to check for table lock leaks"})
136136
public static boolean check_table_lock_leaky = false;
137137

138+
@ConfField(mutable = false, description = {"当前 FE 节点所属的 Resource Group。可通过命令行参数 "
139+
+ "`--local_resource_group` 或环境变量 `DORIS_LOCAL_RESOURCE_GROUP` 覆盖。空字符串表示未设置。",
140+
"The Resource Group that the current FE node belongs to. It can be overridden by the "
141+
+ "`--local_resource_group` command line option or the "
142+
+ "`DORIS_LOCAL_RESOURCE_GROUP` environment variable. An empty string means unset."})
143+
public static String local_resource_group = "";
144+
145+
@ConfField(mutable = false, description = {"是否启用基于 location resource tag 的权限检查。关闭后，不再根据"
146+
+ "用户的 location resource tag 过滤可访问的副本。",
147+
"Whether to enable permission checks based on location resource tags. If disabled, "
148+
+ "accessible replicas will no longer be filtered by the user's location resource tags."})
149+
public static boolean resource_tag_location_check = true;
150+
138151
@ConfField(mutable = true, masterOnly = false,
139152
description = {"PreparedStatement stmtId starting position, used for testing only"})
140153
public static long prepared_stmt_start_id = -1;
@@ -996,6 +1009,14 @@ public class Config extends ConfigBase {
9961009
@ConfField(mutable = true, masterOnly = true)
9971010
public static long tablet_schedule_high_priority_second = 30 * 60;
9981011

1012+
@ConfField(mutable = true, masterOnly = true,
1013+
description = {"是否允许 Resource Group affinity 扩展实现参与副本修复 clone 选源。"
1014+
+ "默认公共实现为 no-op，不改变修复行为。",
1015+
"Whether Resource Group affinity extension implementations may participate in repair clone "
1016+
+ "source selection. The default public implementation is a no-op and does not change repair "
1017+
+ "behavior."})
1018+
public static boolean enable_repair_src_replica_local_affinity = true;
1019+
9991020
/**
10001021
* publish version queue's size in be, report it to fe,
10011022
* if publish task in be exceed direct_publish_limit_number,

‎fe/fe-core/src/main/java/org/apache/doris/DorisFE.java‎

Lines changed: 43 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
import org.apache.doris.qe.QeProcessorImpl;
4343
import org.apache.doris.qe.QeService;
4444
import org.apache.doris.qe.SimpleScheduler;
45+
import org.apache.doris.resource.Tag;
4546
import org.apache.doris.service.ExecuteEnv;
4647
import org.apache.doris.service.FrontendOptions;
4748
import org.apache.doris.tls.server.FeServerStarterFactory;
@@ -89,6 +90,7 @@ public class DorisFE {
8990
private static String LOCK_FILE_PATH;
9091

9192
private static final String LOCK_FILE_NAME = "process.lock";
93+
private static final String LOCAL_RESOURCE_GROUP_ENV = "DORIS_LOCAL_RESOURCE_GROUP";
9294
private static FileChannel processLockFileChannel;
9395
private static FileLock processFileLock;
9496

@@ -136,8 +138,6 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star
136138
return;
137139
}
138140

139-
CommandLineOptions cmdLineOpts = parseArgs(args);
140-
141141
try {
142142
// init config
143143
Config config = new Config();
@@ -146,6 +146,8 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star
146146
// Because the path of custom config file is defined in fe.conf
147147
config.initCustom(Config.custom_config_dir + "/fe_custom.conf");
148148

149+
CommandLineOptions cmdLineOpts = parseArgs(args);
150+
149151
LdapConfig ldapConfig = new LdapConfig();
150152
if (new File(dorisHomeDir + "/conf/ldap.conf").exists()) {
151153
ldapConfig.init(dorisHomeDir + "/conf/ldap.conf");
@@ -382,6 +384,8 @@ private static CommandLineOptions parseArgs(String[] args) {
382384
options.addOption("c", "cluster_snapshot", true, "Specify the cluster snapshot json file");
383385
options.addOption(Option.builder().longOpt(FeConstants.DROP_BACKENDS_KEY)
384386
.desc("When this FE becomes MASTER, drop all backends from cluster metadata (destructive)").build());
387+
options.addOption(null, "local_resource_group", true,
388+
"Specify the local resource group for the current FE");
385389

386390
CommandLine cmd = null;
387391
try {
@@ -392,9 +396,11 @@ private static CommandLineOptions parseArgs(String[] args) {
392396
System.exit(-1);
393397
}
394398

399+
String localResourceGroup = resolveLocalResourceGroup(cmd);
400+
395401
// version
396402
if (cmd.hasOption('v') || cmd.hasOption("version")) {
397-
return new CommandLineOptions(true, "", null, "");
403+
return new CommandLineOptions(true, "", null, "", null, localResourceGroup);
398404
}
399405
// helper
400406
if (cmd.hasOption('h') || cmd.hasOption("helper")) {
@@ -403,7 +409,7 @@ private static CommandLineOptions parseArgs(String[] args) {
403409
System.err.println("Missing helper node");
404410
System.exit(-1);
405411
}
406-
return new CommandLineOptions(false, helperNode, null, "");
412+
return new CommandLineOptions(false, helperNode, null, "", null, localResourceGroup);
407413
}
408414
// image
409415
if (cmd.hasOption('i') || cmd.hasOption("image")) {
@@ -413,7 +419,7 @@ private static CommandLineOptions parseArgs(String[] args) {
413419
System.err.println("imagePath is not set");
414420
System.exit(-1);
415421
}
416-
return new CommandLineOptions(false, "", null, imagePath);
422+
return new CommandLineOptions(false, "", null, imagePath, null, localResourceGroup);
417423
}
418424
if (cmd.hasOption('r') || cmd.hasOption(FeConstants.METADATA_FAILURE_RECOVERY_KEY)) {
419425
System.setProperty(FeConstants.METADATA_FAILURE_RECOVERY_KEY, "true");
@@ -433,7 +439,7 @@ private static CommandLineOptions parseArgs(String[] args) {
433439
if (cmd.hasOption('l') || cmd.hasOption("listdb")) {
434440
// list bdb je databases
435441
BDBToolOptions bdbOpts = new BDBToolOptions(true, "", false, "", "", 0);
436-
return new CommandLineOptions(false, "", bdbOpts, "");
442+
return new CommandLineOptions(false, "", bdbOpts, "", null, localResourceGroup);
437443
}
438444
if (cmd.hasOption('d') || cmd.hasOption("db")) {
439445
// specify a database
@@ -444,7 +450,7 @@ private static CommandLineOptions parseArgs(String[] args) {
444450
}
445451
if (cmd.hasOption('s') || cmd.hasOption("stat")) {
446452
BDBToolOptions bdbOpts = new BDBToolOptions(false, dbName, true, "", "", 0);
447-
return new CommandLineOptions(false, "", bdbOpts, "");
453+
return new CommandLineOptions(false, "", bdbOpts, "", null, localResourceGroup);
448454
}
449455
String fromKey = "";
450456
String endKey = "";
@@ -473,7 +479,7 @@ private static CommandLineOptions parseArgs(String[] args) {
473479
}
474480

475481
BDBToolOptions bdbOpts = new BDBToolOptions(false, dbName, false, fromKey, endKey, metaVersion);
476-
return new CommandLineOptions(false, "", bdbOpts, "");
482+
return new CommandLineOptions(false, "", bdbOpts, "", null, localResourceGroup);
477483

478484
} else {
479485
System.err.println("Invalid options when running bdb je tools");
@@ -487,11 +493,38 @@ private static CommandLineOptions parseArgs(String[] args) {
487493
System.err.println("Missing cluster_snapshot file");
488494
System.exit(-1);
489495
}
490-
return new CommandLineOptions(false, null, null, "", clusterSnapshotFile.trim());
496+
return new CommandLineOptions(false, null, null, "", clusterSnapshotFile.trim(), localResourceGroup);
491497
}
492498

493499
// helper node is null, means no helper node is specified
494-
return new CommandLineOptions(false, null, null, "");
500+
return new CommandLineOptions(false, null, null, "", null, localResourceGroup);
501+
}
502+
503+
private static String resolveLocalResourceGroup(CommandLine cmd) {
504+
String localResourceGroup = Strings.nullToEmpty(Config.local_resource_group);
505+
String source = localResourceGroup.isEmpty() ? "DEFAULT" : "FE_CONF";
506+
if (System.getenv().containsKey(LOCAL_RESOURCE_GROUP_ENV)) {
507+
localResourceGroup = Strings.nullToEmpty(System.getenv(LOCAL_RESOURCE_GROUP_ENV));
508+
source = "ENV";
509+
}
510+
if (cmd.hasOption("local_resource_group")) {
511+
localResourceGroup = Strings.nullToEmpty(cmd.getOptionValue("local_resource_group"));
512+
source = "CMDLINE";
513+
}
514+
515+
LOG.info("effective local_resource_group={}, source={}", localResourceGroup, source);
516+
Config.local_resource_group = localResourceGroup;
517+
if (!localResourceGroup.isEmpty()) {
518+
try {
519+
Tag.create(Tag.TYPE_LOCATION, localResourceGroup);
520+
} catch (Exception e) {
521+
String errMsg = "Invalid local_resource_group: " + localResourceGroup;
522+
LOG.error(errMsg, e);
523+
System.err.println(errMsg);
524+
System.exit(-1);
525+
}
526+
}
527+
return localResourceGroup;
495528
}
496529

497530
private static void printVersion() {

‎fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1241,6 +1241,7 @@ public void initialize(String[] args) throws Exception {
12411241
// 3. Load image first and replay edits
12421242
this.editLog = new EditLog(nodeName);
12431243
loadImage(this.imageDir); // load image file
1244+
seedSelfLocalResourceGroup();
12441245
migrateConstraintsFromTables(); // migrate old table-based constraints
12451246
editLog.open(); // open bdb env
12461247
this.globalTransactionMgr.setEditLog(editLog);
@@ -1271,6 +1272,23 @@ public void initialize(String[] args) throws Exception {
12711272
StmtExecutor.initBlockSqlAstNames();
12721273
}
12731274

1275+
private static void seedSelfLocalResourceGroup() {
1276+
Env env = getCurrentEnv();
1277+
String selfNodeName = env.getNodeName();
1278+
if (Strings.isNullOrEmpty(selfNodeName)) {
1279+
LOG.debug("skip seeding local resource group because self node name is not initialized");
1280+
return;
1281+
}
1282+
1283+
Frontend selfFrontend = env.frontends.get(selfNodeName);
1284+
if (selfFrontend == null) {
1285+
LOG.debug("skip seeding local resource group because self frontend {} is not found", selfNodeName);
1286+
return;
1287+
}
1288+
1289+
selfFrontend.setLocalResourceGroup(Config.local_resource_group);
1290+
}
1291+
12741292
// wait until FE is ready.
12751293
public void waitForReady() throws InterruptedException {
12761294
long counter = 0;
@@ -1391,6 +1409,7 @@ protected void getClusterIdAndRole() throws IOException {
13911409
isFirstTimeStartUp = true;
13921410
Frontend self = new Frontend(role, nodeName, selfNode.getHost(),
13931411
selfNode.getPort());
1412+
self.setLocalResourceGroup(Config.local_resource_group);
13941413
// Set self alive to true, the BDBEnvironment.getReplicationGroupAdmin() will rely on this to get
13951414
// helper node, before the heartbeat thread is started.
13961415
self.setIsAlive(true);
@@ -1874,6 +1893,7 @@ private void transferToMaster() {
18741893

18751894
MetricRepo.init();
18761895

1896+
seedSelfLocalResourceGroup();
18771897
toMasterProgress = "finished";
18781898
canRead.set(true);
18791899
isReady.set(true);

‎fe/fe-core/src/main/java/org/apache/doris/catalog/InternalSchema.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -230,6 +230,10 @@ public class InternalSchema {
230230
ScalarType.createType(PrimitiveType.STRING), ColumnNullableType.NULLABLE));
231231
AUDIT_SCHEMA.add(new ColumnDef("compute_group",
232232
ScalarType.createType(PrimitiveType.STRING), ColumnNullableType.NULLABLE));
233+
AUDIT_SCHEMA.add(new ColumnDef("effective_preferred_resource_group",
234+
ScalarType.createType(PrimitiveType.STRING), ColumnNullableType.NULLABLE));
235+
AUDIT_SCHEMA.add(new ColumnDef("resource_group_select_policy",
236+
ScalarType.createType(PrimitiveType.STRING), ColumnNullableType.NULLABLE));
233237
// Keep stmt as last column. So that in fe.audit.log, it will be easier to get sql string
234238
AUDIT_SCHEMA.add(new ColumnDef("stmt",
235239
ScalarType.createType(PrimitiveType.STRING), ColumnNullableType.NULLABLE));

‎fe/fe-core/src/main/java/org/apache/doris/clone/TabletSchedCtx.java‎

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,8 @@
4040
import org.apache.doris.common.util.DebugUtil;
4141
import org.apache.doris.common.util.TimeUtils;
4242
import org.apache.doris.persist.ReplicaPersistInfo;
43+
import org.apache.doris.resource.ResourceGroupAffinityPolicy;
44+
import org.apache.doris.resource.ResourceGroupAffinityPolicyFactory;
4345
import org.apache.doris.resource.Tag;
4446
import org.apache.doris.system.Backend;
4547
import org.apache.doris.system.SystemInfoService;
@@ -194,6 +196,9 @@ public enum State {
194196
// tag is only set for BALANCE task, used to identify which workload group this Balance job is in
195197
private Tag tag;
196198

199+
private ResourceGroupAffinityPolicy.SrcAffinityResult srcAffinityResult =
200+
ResourceGroupAffinityPolicy.SrcAffinityResult.DISABLED;
201+
197202
private SubCode schedFailedCode;
198203

199204
private boolean isUniqKeyMergeOnWrite = false;
@@ -446,6 +451,10 @@ public long getCopyTimeMs() {
446451
return copyTimeMs;
447452
}
448453

454+
public ResourceGroupAffinityPolicy.SrcAffinityResult getSrcAffinityResult() {
455+
return srcAffinityResult;
456+
}
457+
449458
public long getSrcBackendId() {
450459
if (srcReplica != null) {
451460
return srcReplica.getBackendIdWithoutException();
@@ -653,7 +662,15 @@ public void chooseSrcReplica(Map<Long, PathSlot> backendsWorkingSlots, long exce
653662
// choose a replica which slot is available from candidates.
654663
// sort replica by version count asc and isUserDrop, so that we prefer to choose replicas with fewer versions
655664
Collections.sort(candidates, CLONE_SRC_COMPARATOR);
656-
for (Replica srcReplica : candidates) {
665+
666+
// Resource-group (tag.location) source affinity is behind an SPI. The default no-op policy
667+
// leaves the candidate ordering unchanged.
668+
ResourceGroupAffinityPolicy affinityPolicy = ResourceGroupAffinityPolicyFactory.get();
669+
boolean affinity = isRepairSrcAffinityActive(affinityPolicy, destBackendId);
670+
List<Replica> orderedCandidates = affinity
671+
? affinityPolicy.orderRepairSrcCandidates(candidates, destBackendId)
672+
: candidates;
673+
for (Replica srcReplica : orderedCandidates) {
657674
long replicaBeId = srcReplica.getBackendIdWithoutException();
658675
PathSlot slot = backendsWorkingSlots.get(replicaBeId);
659676
if (slot == null) {
@@ -673,12 +690,22 @@ public void chooseSrcReplica(Map<Long, PathSlot> backendsWorkingSlots, long exce
673690
continue;
674691
}
675692
setSrc(srcReplica);
693+
srcAffinityResult = affinity
694+
? affinityPolicy.classifyRepairSrc(replicaBeId, destBackendId,
695+
tablet.getReplicas(), candidates)
696+
: ResourceGroupAffinityPolicy.SrcAffinityResult.DISABLED;
676697
return;
677698
}
678699
throw new SchedException(Status.SCHEDULE_FAILED, SubCode.WAITING_SLOT,
679700
"waiting for source replica's slot");
680701
}
681702

703+
static boolean isRepairSrcAffinityActive(ResourceGroupAffinityPolicy affinityPolicy, long destBackendId) {
704+
return Config.enable_repair_src_replica_local_affinity
705+
&& affinityPolicy.isRepairSrcAffinityEnabled()
706+
&& destBackendId != -1;
707+
}
708+
682709
/*
683710
* Same rules as choosing source replica for supplement.
684711
* But we need to check that we can not choose the same replica as dest replica,

0 commit comments

Comments
 (0)