This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new cb5b9cb69a1 [fix](fe) Check the workload group ALTER ROUTINE LOAD
sets, as CREATE does (#70130)
cb5b9cb69a1 is described below
commit cb5b9cb69a1a1c4454e9439a6eabd8dd4a35b6c6
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Thu Nov 5 08:19:04 2026 +0000
[fix](fe) Check the workload group ALTER ROUTINE LOAD sets, as CREATE does
(#70130)
### What problem does this PR solve?
Issue Number: None
Related PR: #69749, #50563, #49960, #53490
Problem Summary:
**In short.** `ALTER ROUTINE LOAD ... PROPERTIES("workload_group" =
"...")` accepts any name. `CREATE ROUTINE LOAD` checks the workload
group as the job's owner in the compute group the job runs in, and every
task of the job resolves it the same way, so a name that is missing
there, or that the owner may not use, is accepted by the ALTER and then
fails every task. This PR checks it at ALTER time, as CREATE does.
**Background.** A workload group belongs to a compute group in cloud
mode, and to the owner's resource group outside it. Each task looks the
job's workload group up as the owner in the job's compute group
(`KafkaTaskInfo#rePlan`, `KinesisTaskInfo#rePlan`). Kafka jobs used to
check an altered workload group (#50563, in the `AlterRoutineLoadStmt`
path). The Nereids `AlterRoutineLoadCommand` path (#49960) did not carry
that check, and #53490 removed the old path. #69749 brought a check back
only for an ALTER that moves a Kafka job to another compute group. The
comment that delays the check ("we can only get clusterId when alter
job") is no longer true: the ALTER has the job at hand. Found while
reviewing #69749 (thread 4415694066).
**The problem, and what it cost.** Measured on master `856194ece0e` with
unit tests; each ALTER below goes straight to the PAUSED check, so on a
paused job it is accepted and written to the edit log:
| Job | ALTER | Result |
|---|---|---|
| Kafka, cloud mode, runs in `cg_a` | `"workload_group" = "wg_b"`, which
only `cg_b` has | accepted |
| Kafka, outside cloud mode | a workload group that does not exist |
accepted |
| Kinesis, outside cloud mode | a workload group that does not exist |
accepted |
After RESUME, the first task of each such job fails with `Can not find
workload group ...`, and the job pauses with `CREATE_TASKS_ERR`. That
code is retryable, so auto resume repeats this with backoff (20 s up to
300 s) for as long as the job exists. The ALTER that caused it reported
success.
**How this PR fixes it.**
- `RoutineLoadJob#checkAlteredWorkloadGroup`, called by the Kafka and
the Kinesis job first thing under the job's write lock, before the
PAUSED check, so that of two ALTERs of one job running at once the later
one checks what the earlier one left. The Kafka offset conversion stays
before the lock, so an ALTER with Kafka offsets and a workload group it
cannot use converts the offsets before it fails. When an ALTER sets a
workload group or declares a compute group, it checks the resulting
workload group in the resulting compute group, with
`CreateRoutineLoadInfo.checkWorkloadGroup` as CREATE does. The resulting
workload group is the one the ALTER sets, else the one the job keeps;
the resulting compute group is the one the ALTER declares, else the
job's. The check runs as the owner, because the tasks run as the owner.
A rejected ALTER changes nothing and writes no edit log. The replay of
an ALTER does not go through this path.
- The Kafka job's check for a move to another compute group (#69749) is
replaced by this one; it checks the same thing for that ALTER.
- The comments in `AlterRoutineLoadCommand` that pointed at the old
place, or at the old reason for not checking, are corrected.
- `test_routine_load_alter` altered a job to `alter_routine_load_group`
without creating the group: it worked only because the ALTER did not
check, or because `test_routine_load` happened to create the group
first. It now creates the group, the way `test_routine_load` does, and
checks that an ALTER to a missing group fails.
**Results.**
| ALTER | Before | After |
|---|---|---|
| a workload group the job's compute group has, and the owner may use |
accepted | accepted |
| a workload group the job's compute group lacks, or the owner may not
use | accepted, then every task fails and auto resume loops | rejected,
e.g. `Can not find workload group wg_b in compute croup cg_a`; the job
is unchanged |
| a compute group, keeping a workload group the new group lacks (Kafka)
| rejected (#69749) | rejected |
| neither property | not checked | not checked |
branch-4.1 and branch-4.2 have the same gap.
---
.../doris/load/routineload/RoutineLoadJob.java | 23 +++
.../routineload/kafka/KafkaRoutineLoadJob.java | 12 +-
.../routineload/kinesis/KinesisRoutineLoadJob.java | 1 +
.../plans/commands/AlterRoutineLoadCommand.java | 6 +-
.../routineload/RoutineLoadComputeGroupTest.java | 177 +++++++++++++++++++--
.../routine_load/test_routine_load_alter.groovy | 13 ++
6 files changed, 201 insertions(+), 31 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
index d861bd6a4c8..49a97421401 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
@@ -2169,6 +2169,29 @@ public abstract class RoutineLoadJob
public abstract void modifyProperties(AlterRoutineLoadCommand command)
throws UserException;
+ /**
+ * Checks the workload group an ALTER leaves the job with, in the compute
group the ALTER leaves the job in, as
+ * CREATE does and as every task resolves it ({@code
KafkaTaskInfo#rePlan}): as the owner, whom the tasks run
+ * as. One that group does not have, or the owner may not use, would
otherwise fail every task with the
+ * retryable CREATE_TASKS_ERR, which auto resume repeats for good. Only an
ALTER that sets a workload group or
+ * declares a compute group changes what the tasks look up; a workload
group it does not set is the one the
+ * job keeps. Called on the statement thread under the job's write lock,
so that of two ALTERs running at once
+ * the later one checks what the earlier one left; never when an ALTER is
replayed.
+ */
+ protected void checkAlteredWorkloadGroup(Map<String, String>
alteredJobProperties) throws UserException {
+ String declaredComputeGroup = alteredJobProperties.get(COMPUTE_GROUP);
+ String alteredWorkloadGroup = alteredJobProperties.get(WORKLOAD_GROUP);
+ if (StringUtils.isEmpty(declaredComputeGroup) &&
StringUtils.isEmpty(alteredWorkloadGroup)) {
+ return;
+ }
+ String workloadGroup = StringUtils.isEmpty(alteredWorkloadGroup) ?
getWorkloadGroup() : alteredWorkloadGroup;
+ if (StringUtils.isEmpty(workloadGroup)) {
+ return;
+ }
+ CreateRoutineLoadInfo.checkWorkloadGroup(getUserIdentity(),
workloadGroup,
+ StringUtils.isEmpty(declaredComputeGroup) ? getCloudCluster()
: declaredComputeGroup);
+ }
+
public abstract void
replayModifyProperties(AlterRoutineLoadJobOperationLog log);
public abstract NereidsRoutineLoadTaskInfo toNereidsRoutineLoadTaskInfo()
throws UserException;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
index 8477c6ed7ed..8e6a9fb745a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
@@ -797,17 +797,6 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
Map<String, String> jobProperties = command.getAnalyzedJobProperties();
KafkaDataSourceProperties dataSourceProperties =
(KafkaDataSourceProperties) command.getDataSourceProperties();
String declaredComputeGroup =
jobProperties.get(CreateRoutineLoadInfo.COMPUTE_GROUP);
- if (!Strings.isNullOrEmpty(declaredComputeGroup)) {
- // Every task looks the job's workload group up in the compute
group the job runs in
- // (KafkaTaskInfo#rePlan), so moving the job checks it in the new
group, as CREATE does in the group
- // it declares. One the new group does not have would otherwise
fail every task with the retryable
- // CREATE_TASKS_ERR, which auto resume repeats for good. Checked
as the owner, whom the tasks run as.
- String workloadGroup =
jobProperties.containsKey(CreateRoutineLoadInfo.WORKLOAD_GROUP)
- ? jobProperties.get(CreateRoutineLoadInfo.WORKLOAD_GROUP)
: getWorkloadGroup();
- if (!Strings.isNullOrEmpty(workloadGroup)) {
- CreateRoutineLoadInfo.checkWorkloadGroup(getUserIdentity(),
workloadGroup, declaredComputeGroup);
- }
- }
if (null != dataSourceProperties) {
// The offsets are converted before modifyPropertiesInternal
installs the new properties,
// so a compute group this ALTER declares has to be used already:
for
@@ -820,6 +809,7 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
writeLock();
try {
+ checkAlteredWorkloadGroup(jobProperties);
if (getState() != JobState.PAUSED) {
throw new DdlException("Only supports modification of PAUSED
jobs");
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java
index 2a9e566a80d..7dec58a33cd 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java
@@ -678,6 +678,7 @@ public class KinesisRoutineLoadJob extends RoutineLoadJob {
writeLock();
try {
+ checkAlteredWorkloadGroup(jobProperties);
if (getState() != JobState.PAUSED) {
throw new DdlException("Only supports modification of PAUSED
jobs");
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java
index 4e52c730cab..8196c1c1aff 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java
@@ -293,7 +293,7 @@ public class AlterRoutineLoadCommand extends AlterCommand {
String computeGroup =
jobProperties.get(CreateRoutineLoadInfo.COMPUTE_GROUP);
if (!StringUtil.isEmpty(computeGroup)) {
// Validated right here instead of letting a bad name pause
the whole job on the next task.
- // The job's workload group is then checked in this group by
KafkaRoutineLoadJob#modifyProperties,
+ // The job's workload group is then checked in this group by
RoutineLoadJob#checkAlteredWorkloadGroup,
// which also knows the one the job keeps when this ALTER sets
none.
ComputeGroupBindingUtil.validateDeclaredComputeGroup(ctx,
computeGroup);
// a combined ALTER converts its new Kafka offsets in this
group, see KafkaRoutineLoadJob
@@ -305,8 +305,8 @@ public class AlterRoutineLoadCommand extends AlterCommand {
if (jobProperties.containsKey(CreateRoutineLoadInfo.WORKLOAD_GROUP)) {
String workloadGroup =
jobProperties.get(CreateRoutineLoadInfo.WORKLOAD_GROUP);
if (!StringUtil.isEmpty(workloadGroup)) {
- // NOTE: delay check workload group's existence check when
alter routine load job
- // because we can only get clusterId when alter job.
+ // Checked by RoutineLoadJob#checkAlteredWorkloadGroup, in the
compute group of the job and as its
+ // owner, which only the job knows.
analyzedJobProperties.put(CreateRoutineLoadInfo.WORKLOAD_GROUP,
jobProperties.get(CreateRoutineLoadInfo.WORKLOAD_GROUP));
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadComputeGroupTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadComputeGroupTest.java
index e167da1bc8a..c71f303ad43 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadComputeGroupTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadComputeGroupTest.java
@@ -43,6 +43,7 @@ import
org.apache.doris.nereids.trees.plans.commands.AlterRoutineLoadCommand;
import
org.apache.doris.nereids.trees.plans.commands.info.CreateRoutineLoadInfo;
import org.apache.doris.nereids.trees.plans.commands.info.LabelNameInfo;
import org.apache.doris.nereids.trees.plans.commands.load.LoadProperty;
+import org.apache.doris.persist.EditLog;
import org.apache.doris.persist.gson.GsonUtils;
import org.apache.doris.persist.meta.PersistMetaModules;
import org.apache.doris.qe.ConnectContext;
@@ -69,7 +70,9 @@ import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;
+import java.util.function.Consumer;
import java.util.stream.Collectors;
/**
@@ -91,6 +94,7 @@ public class RoutineLoadComputeGroupTest {
private AccessControllerManager originalAccessManager;
private WorkloadGroupMgr originalWorkloadGroupMgr;
private ComputeGroupMgr originalComputeGroupMgr;
+ private EditLog originalEditLog;
@BeforeEach
public void setUp() {
@@ -100,6 +104,7 @@ public class RoutineLoadComputeGroupTest {
originalAccessManager = Env.getCurrentEnv().getAccessManager();
originalWorkloadGroupMgr = Env.getCurrentEnv().getWorkloadGroupMgr();
originalComputeGroupMgr = Env.getCurrentEnv().getComputeGroupMgr();
+ originalEditLog = Env.getCurrentEnv().getEditLog();
}
@AfterEach
@@ -110,6 +115,7 @@ public class RoutineLoadComputeGroupTest {
Deencapsulation.setField(Env.getCurrentEnv(), "accessManager",
originalAccessManager);
Deencapsulation.setField(Env.getCurrentEnv(), "workloadGroupMgr",
originalWorkloadGroupMgr);
Deencapsulation.setField(Env.getCurrentEnv(), "computeGroupMgr",
originalComputeGroupMgr);
+ Deencapsulation.setField(Env.getCurrentEnv(), "editLog",
originalEditLog);
ConnectContext.remove();
}
@@ -499,21 +505,7 @@ public class RoutineLoadComputeGroupTest {
public void testAlterChecksWorkloadGroupInTheComputeGroupItDeclares()
throws UserException {
enterCloudMode(true, Lists.newArrayList("cg_a", "cg_b"));
// only cg_a has wg_a, only cg_b has wg_b
- List<String> lookups = Lists.newArrayList();
- WorkloadGroupMgr workloadGroupMgr =
Mockito.mock(WorkloadGroupMgr.class);
-
Mockito.when(workloadGroupMgr.getWorkloadGroup(Mockito.any(ConnectContext.class))).thenAnswer(invocation
-> {
- ConnectContext lookupCtx = invocation.getArgument(0);
- String computeGroup =
lookupCtx.getSessionVariable().getCloudCluster();
- String workloadGroup =
lookupCtx.getSessionVariable().getWorkloadGroup();
- String lookup = computeGroup + "/" + workloadGroup;
- lookups.add(lookupCtx.getCurrentUserIdentity().getQualifiedUser()
+ ":" + lookup);
- if (!"cg_a/wg_a".equals(lookup) && !"cg_b/wg_b".equals(lookup)) {
- throw new UserException("Can not find workload group " +
workloadGroup + " in compute croup "
- + computeGroup);
- }
- return Lists.newArrayList(Mockito.mock(WorkloadGroup.class));
- });
- Deencapsulation.setField(Env.getCurrentEnv(), "workloadGroupMgr",
workloadGroupMgr);
+ List<String> lookups = mockWorkloadGroups("cg_a/wg_a", "cg_b/wg_b");
KafkaRoutineLoadJob job = newJob(SESSION_SNAPSHOT_CG, "cg_a");
Map<String, String> jobProperties = Deencapsulation.getField(job,
"jobProperties");
@@ -536,13 +528,164 @@ public class RoutineLoadComputeGroupTest {
Assertions.assertTrue(notPaused.getMessage().contains("Only supports
modification of PAUSED jobs"),
notPaused.getMessage());
- // an ALTER that declares no compute group looks nothing up, as before
+ // an ALTER that sets neither looks nothing up
Assertions.assertThrows(DdlException.class,
- () -> job.modifyProperties(newJobPropertiesAlter(null,
"wg_unknown")));
+ () -> job.modifyProperties(newJobPropertiesAlter(null, null)));
Assertions.assertEquals(Lists.newArrayList("owner:cg_b/wg_a",
"owner:cg_b/wg_b"), lookups);
}
+ /**
+ * An ALTER that only sets a workload group checks it in the compute group
the job runs in, as the owner, as
+ * CREATE does and as every task resolves it ({@code
KafkaTaskInfo#rePlan}).
+ *
+ * <p>It used to be adopted unchecked. A job running in cg_a, ALTERed to
("workload_group" = "wg_b") where only
+ * cg_b has wg_b, then failed every task with "Can not find workload group
wg_b in compute croup cg_a" and
+ * paused with the retryable CREATE_TASKS_ERR, which auto resume repeated
for good.
+ */
+ @Test
+ public void testAlterOfWorkloadGroupChecksItInTheComputeGroupOfTheJob()
throws UserException {
+ enterCloudMode(true, Lists.newArrayList("cg_a", "cg_b"));
+ // only cg_a has wg_a, only cg_b has wg_b
+ List<String> lookups = mockWorkloadGroups("cg_a/wg_a", "cg_b/wg_b");
+ UserIdentity owner =
UserIdentity.createAnalyzedUserIdentWithIp("owner", "%");
+ // runs in cg_a, the group of the session that created it
+ KafkaRoutineLoadJob job = newJob("cg_a", null);
+ // runs in cg_b, the group it declares
+ KafkaRoutineLoadJob declaringJob = newJob("cg_a", "cg_b");
+ for (KafkaRoutineLoadJob j : Lists.newArrayList(job, declaringJob)) {
+ Deencapsulation.setField(j, "userIdentity", owner);
+ // left RUNNING, so an ALTER that gets past the check stops at the
PAUSED check right after it
+ Deencapsulation.setField(j, "state",
RoutineLoadJob.JobState.RUNNING);
+ }
+
+ UserException e = Assertions.assertThrows(UserException.class,
+ () -> job.modifyProperties(newJobPropertiesAlter(null,
"wg_b")));
+ Assertions.assertTrue(e.getMessage().contains("Can not find workload
group wg_b in compute croup cg_a"),
+ e.getMessage());
+ DdlException notPaused = Assertions.assertThrows(DdlException.class,
+ () -> job.modifyProperties(newJobPropertiesAlter(null,
"wg_a")));
+ Assertions.assertTrue(notPaused.getMessage().contains("Only supports
modification of PAUSED jobs"),
+ notPaused.getMessage());
+
+ e = Assertions.assertThrows(UserException.class,
+ () ->
declaringJob.modifyProperties(newJobPropertiesAlter(null, "wg_a")));
+ Assertions.assertTrue(e.getMessage().contains("Can not find workload
group wg_a in compute croup cg_b"),
+ e.getMessage());
+
+ Assertions.assertEquals(Lists.newArrayList("owner:cg_a/wg_b",
"owner:cg_a/wg_a", "owner:cg_b/wg_a"), lookups);
+ }
+
+ /**
+ * Outside cloud mode a workload group lives in the namespace the owner's
resource tag picks, which the lookup
+ * resolves from the owner alone; both Kafka and Kinesis jobs check an
altered one there.
+ */
+ @Test
+ public void testAlterOfWorkloadGroupChecksItOutsideCloudMode() throws
UserException {
+ List<String> lookups = mockWorkloadGroups("/wg_a");
+ UserIdentity owner =
UserIdentity.createAnalyzedUserIdentWithIp("owner", "%");
+ KafkaRoutineLoadJob kafkaJob = newJob(null, null);
+ Deencapsulation.setField(kafkaJob, "userIdentity", owner);
+ KinesisRoutineLoadJob kinesisJob = new KinesisRoutineLoadJob(1L,
"kinesis_routine_load_job", 1L, 1L,
+ "ap-southeast-1", "stream-1", owner);
+ for (RoutineLoadJob job : Lists.<RoutineLoadJob>newArrayList(kafkaJob,
kinesisJob)) {
+ Deencapsulation.setField(job, "state",
RoutineLoadJob.JobState.RUNNING);
+ UserException e = Assertions.assertThrows(UserException.class,
+ () -> job.modifyProperties(newJobPropertiesAlter(null,
"wg_unknown")));
+ Assertions.assertTrue(e.getMessage().contains("Can not find
workload group wg_unknown"), e.getMessage());
+ DdlException notPaused =
Assertions.assertThrows(DdlException.class,
+ () -> job.modifyProperties(newJobPropertiesAlter(null,
"wg_a")));
+ Assertions.assertTrue(notPaused.getMessage().contains("Only
supports modification of PAUSED jobs"),
+ notPaused.getMessage());
+ }
+
+ Assertions.assertEquals(Lists.newArrayList("owner:/wg_unknown",
"owner:/wg_a", "owner:/wg_unknown",
+ "owner:/wg_a"), lookups);
+ }
+
+ /**
+ * Of two ALTERs of one paused job that run at once, the later one checks
what the earlier one left.
+ *
+ * <p>A job in cg_a with wg_old, which both cg_a and cg_b have. One ALTER
moves it to cg_b; another, run while
+ * the first looks wg_old up in cg_b, sets wg_a, which only cg_a has. When
the check ran before the job lock,
+ * the second one checked wg_a in cg_a, both were accepted, and the job
was left with wg_a in cg_b, which fails
+ * every task with "Can not find workload group wg_a in compute croup
cg_b".
+ */
+ @Test
+ public void testConcurrentAltersCheckWhatTheEarlierOneLeft() throws
Exception {
+ enterCloudMode(true, Lists.newArrayList("cg_a", "cg_b"));
+ KafkaRoutineLoadJob job = newJob("cg_a", null);
+ Map<String, String> jobProperties = Deencapsulation.getField(job,
"jobProperties");
+ jobProperties.put(RoutineLoadJob.WORKLOAD_GROUP, "wg_old");
+ Deencapsulation.setField(job, "userIdentity",
UserIdentity.createAnalyzedUserIdentWithIp("owner", "%"));
+ Deencapsulation.setField(job, "state", RoutineLoadJob.JobState.PAUSED);
+ EditLog editLog = Mockito.mock(EditLog.class);
+ Deencapsulation.setField(Env.getCurrentEnv(), "editLog", editLog);
+
+ AlterRoutineLoadCommand setWorkloadGroup = newJobPropertiesAlter(null,
"wg_a");
+ AtomicReference<Exception> otherAlterFailure = new AtomicReference<>();
+ Thread otherAlter = new Thread(() -> {
+ try {
+ job.modifyProperties(setWorkloadGroup);
+ } catch (Exception e) {
+ otherAlterFailure.set(e);
+ }
+ });
+ // only cg_a has wg_a. The other ALTER starts when the move to cg_b
looks wg_old up in cg_b, and that lookup
+ // returns once the other ALTER has finished or waits for the job lock.
+ List<String> lookups = mockWorkloadGroups(lookup -> {
+ if (!lookup.equals("owner:cg_b/wg_old")) {
+ return;
+ }
+ otherAlter.start();
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (otherAlter.isAlive() &&
!job.lock.hasQueuedThread(otherAlter) && System.nanoTime() < deadline) {
+ LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1));
+ }
+ }, "cg_a/wg_old", "cg_b/wg_old", "cg_a/wg_a");
+
+ job.modifyProperties(newJobPropertiesAlter("cg_b", null));
+ otherAlter.join();
+
+ Assertions.assertNotNull(otherAlterFailure.get(), "the later ALTER
must check wg_a in cg_b");
+ Assertions.assertTrue(otherAlterFailure.get().getMessage().contains(
+ "Can not find workload group wg_a in compute croup cg_b"),
otherAlterFailure.get().getMessage());
+ Assertions.assertEquals("cg_b", job.getCloudCluster());
+ Assertions.assertEquals("wg_old", job.getWorkloadGroup());
+ Assertions.assertEquals(Lists.newArrayList("owner:cg_b/wg_old",
"owner:cg_b/wg_a"), lookups);
+ Mockito.verify(editLog,
Mockito.times(1)).logAlterRoutineLoadJob(Mockito.any());
+ }
+
+ // Installs a WorkloadGroupMgr that has only these workload groups, each
written "<compute group>/<workload
+ // group>", with an empty compute group outside cloud mode, where a lookup
names none. Returns every lookup in
+ // call order, written "<user>:<compute group>/<workload group>".
+ private List<String> mockWorkloadGroups(String... existing) throws
UserException {
+ return mockWorkloadGroups(lookup -> { }, existing);
+ }
+
+ // The same, handing each lookup to onLookup before answering it.
+ private List<String> mockWorkloadGroups(Consumer<String> onLookup,
String... existing) throws UserException {
+ List<String> existingGroups = Lists.newArrayList(existing);
+ List<String> lookups = Lists.newArrayList();
+ WorkloadGroupMgr workloadGroupMgr =
Mockito.mock(WorkloadGroupMgr.class);
+
Mockito.when(workloadGroupMgr.getWorkloadGroup(Mockito.any(ConnectContext.class))).thenAnswer(invocation
-> {
+ ConnectContext lookupCtx = invocation.getArgument(0);
+ String computeGroup =
lookupCtx.getSessionVariable().getCloudCluster();
+ String workloadGroup =
lookupCtx.getSessionVariable().getWorkloadGroup();
+ String lookup = computeGroup + "/" + workloadGroup;
+ String recorded =
lookupCtx.getCurrentUserIdentity().getQualifiedUser() + ":" + lookup;
+ lookups.add(recorded);
+ onLookup.accept(recorded);
+ if (!existingGroups.contains(lookup)) {
+ throw new UserException("Can not find workload group " +
workloadGroup + " in compute croup "
+ + computeGroup);
+ }
+ return Lists.newArrayList(Mockito.mock(WorkloadGroup.class));
+ });
+ Deencapsulation.setField(Env.getCurrentEnv(), "workloadGroupMgr",
workloadGroupMgr);
+ return lookups;
+ }
+
// What AlterRoutineLoadCommand#validate leaves behind for
PROPERTIES("compute_group" = ...,
// "workload_group" = ...), with each property only when it is given.
private AlterRoutineLoadCommand newJobPropertiesAlter(String computeGroup,
String workloadGroup) {
diff --git
a/regression-test/suites/load_p0/routine_load/test_routine_load_alter.groovy
b/regression-test/suites/load_p0/routine_load/test_routine_load_alter.groovy
index 32571b5e29a..3b707872c42 100644
--- a/regression-test/suites/load_p0/routine_load/test_routine_load_alter.groovy
+++ b/regression-test/suites/load_p0/routine_load/test_routine_load_alter.groovy
@@ -227,6 +227,15 @@ suite("test_routine_load_alter","p0") {
}
// test show after alter
+ // ALTER checks the workload group like CREATE does, so the one
altered to below has to exist
+ def forComputeGroupStr = ""
+ if (isCloudMode()) {
+ def clusters = sql " SHOW CLUSTERS; "
+ assertTrue(!clusters.isEmpty())
+ def validCluster = clusters[0][0]
+ forComputeGroupStr = " for $validCluster "
+ }
+ sql "create workload group if not exists alter_routine_load_group
$forComputeGroupStr properties ( 'min_cpu_percent'='0');"
try {
sql """
CREATE ROUTINE LOAD ${jobName} ON ${tableName}
@@ -257,6 +266,10 @@ suite("test_routine_load_alter","p0") {
sql "ALTER ROUTINE LOAD FOR ${jobName} PROPERTIES(\"timezone\" =
\"Asia/Shanghai\");"
sql "ALTER ROUTINE LOAD FOR ${jobName}
PROPERTIES(\"num_as_string\" = \"true\");"
sql "ALTER ROUTINE LOAD FOR ${jobName} PROPERTIES(\"fuzzy_parse\"
= \"true\");"
+ test {
+ sql "ALTER ROUTINE LOAD FOR ${jobName}
PROPERTIES(\"workload_group\" = \"test_routine_load_alter_no_such_group\");"
+ exception "Can not find workload group
test_routine_load_alter_no_such_group"
+ }
sql "ALTER ROUTINE LOAD FOR ${jobName}
PROPERTIES(\"workload_group\" = \"alter_routine_load_group\");"
sql "ALTER ROUTINE LOAD FOR ${jobName}
PROPERTIES(\"max_filter_ratio\" = \"0.5\");"
sql "ALTER ROUTINE LOAD FOR ${jobName} COLUMNS(k00, k01, k02, k03,
k04, k05);"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
