morningman commented on code in PR #69749:
URL: https://github.com/apache/doris/pull/69749#discussion_r4401394850
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java:
##########
@@ -599,13 +625,30 @@ public void checkJobProperties() throws UserException {
RoutineLoadJob.DEFAULT_LOAD_TO_SINGLE_TABLET,
LOAD_TO_SINGLE_TABLET + " should be a boolean");
+ String inputComputeGroupStr = jobProperties.get(COMPUTE_GROUP);
+ if (!StringUtils.isEmpty(inputComputeGroupStr)) {
+ // The name is always adopted, because the workload group check
below resolves in its
+ // namespace, but it is only validated when a user is actually
declaring it. On the
+ // metadata load path the declared group may have been dropped
long ago, and failing
+ // there would cancel the job permanently instead of pausing it -
see isReplay.
+ if (!isReplay) {
+
ComputeGroupBindingUtil.validateDeclaredComputeGroup(ConnectContext.get(),
inputComputeGroupStr);
+ }
+ this.computeGroupName = inputComputeGroupStr;
+ }
+
String inputWorkloadGroupStr = jobProperties.get(WORKLOAD_GROUP);
if (!StringUtils.isEmpty(inputWorkloadGroupStr)) {
ConnectContext tmpCtx = new ConnectContext();
tmpCtx.setCurrentUserIdentity(ConnectContext.get().getCurrentUserIdentity());
tmpCtx.getSessionVariable().setWorkloadGroup(inputWorkloadGroupStr);
if (Config.isCloudMode()) {
- tmpCtx.setCloudCluster(ConnectContext.get().getCloudCluster());
+ // A workload group lives in the namespace of a compute group:
the same name under a
+ // different compute group is a different group. So this
existence check must be done
+ // against the compute group the job will actually run in, not
the session's one.
+
tmpCtx.setCloudCluster(StringUtils.isEmpty(this.computeGroupName)
+ ? ConnectContext.get().getCloudCluster()
+ : this.computeGroupName);
Review Comment:
Confirmed, and fixed in 4ed5856f57c. On the metadata load path
`setReplay(true)` skipped the compute group check, but the workload group
existence check right after it still resolved the declared group
(`WorkloadGroupMgr.getWorkloadGroup` -> `ConnectContext.getComputeGroup()` ->
`ComputeGroupMgr.getComputeGroupByName`), which throws once that group is
dropped, renamed or scaled to zero, and `gsonPostProcess` turns the throw into
the final `CANCELLED`. It is easier to hit than a dropped group alone:
`origStmt` keeps the CREATE-time `compute_group`, so a job moved to another
group by ALTER was cancelled by the next restart after its old group was
dropped.
The same lookup also failed on image load without any compute group.
`PersistMetaModules` reads `routineLoadJobs` before `workloadGroups`, and until
`Env.loadWorkloadGroups` runs the Env holds the empty `WorkloadGroupMgr` its
constructor created, so a job created with `workload_group` could not find its
group there (`Can not find workload group wg_a in compute group default`) and
was cancelled by the first restart that loaded it from an image. That part
predates this PR.
The re-parse only rebuilds the `RoutineLoadDesc` and never uses the resolved
workload group, so the replay path now adopts both names without resolving
either; CREATE and ALTER still validate them.
`testMetadataLoadDoesNotResolveWorkloadGroupInDeclaredComputeGroup` replays a
statement that declares a dropped compute group together with a workload group,
and `testImageLoadAdoptsWorkloadGroupBeforeWorkloadGroupsAreLoaded` replays one
with only a workload group against an empty `WorkloadGroupMgr`, as on image
load; both fail without the change. branch-4.x has the same code.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java:
##########
@@ -438,6 +439,13 @@ protected void setOptional(CreateRoutineLoadInfo info)
throws UserException {
if (!StringUtils.isEmpty(info.getWorkloadGroupName())) {
jobProperties.put(WORKLOAD_GROUP, info.getWorkloadGroupName());
}
+
+ // Only write the key when the user declared it. An absent key means
"not declared" and
+ // keeps the existing implicit resolution, which is also what later
versions rely on to
+ // tell a pinned group apart from an inherited one.
+ if (!StringUtils.isEmpty(info.getComputeGroupName())) {
+ jobProperties.put(COMPUTE_GROUP, info.getComputeGroupName());
Review Comment:
Confirmed for CREATE and ALTER, and fixed in 4ed5856f57c. After the USAGE
and existence checks, the declared group is now started with
`CloudSystemInfoService.waitForAutoStart`
(`ComputeGroupBindingUtil.activateDeclaredComputeGroup`), the third step `USE
@<group>` takes in `CloudEnv.changeCloudCluster`. It runs on the statement
thread before the job lock or any Kafka metadata request, does not change the
caller's session group, and is not called on the metadata load path. It is not
part of `validateDeclaredComputeGroup`, which async MVs share: an MV sends
nothing to its group at CREATE, and its refresh resolves backends through
`getBackendsByCurrentCluster`, which already starts the group it binds.
The scheduler half is not specific to declared groups.
`CloudRoutineLoadManager.getAvailableBackendIds` and
`KafkaUtil.getBackendIdsForMetaRequest` resolve backends by name and never
start a suspended group on master, so a job whose implicit group is suspended
after creation already pauses and retries without resuming it. Changing that
affects every cloud routine load job and is out of scope for this port.
`testCreateWakesDeclaredComputeGroup` and `testAlterWakesDeclaredComputeGroup`
cover the two call sites, and the metadata load test checks that replay starts
nothing.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java:
##########
@@ -287,6 +289,16 @@ private void checkJobProperties() throws UserException {
analyzedJobProperties.put(CreateRoutineLoadInfo.UNIQUE_KEY_UPDATE_MODE,
modeStr.toUpperCase());
}
+ if (jobProperties.containsKey(CreateRoutineLoadInfo.COMPUTE_GROUP)) {
+ String computeGroup =
jobProperties.get(CreateRoutineLoadInfo.COMPUTE_GROUP);
+ if (!StringUtil.isEmpty(computeGroup)) {
+ // Unlike workload group, the compute group can be fully
validated right here, so do
+ // it now instead of letting a bad name pause the whole job on
the next task.
+ ComputeGroupBindingUtil.validateDeclaredComputeGroup(ctx,
computeGroup);
+ analyzedJobProperties.put(CreateRoutineLoadInfo.COMPUTE_GROUP,
computeGroup);
Review Comment:
Confirmed, and fixed in 4ed5856f57c. `KafkaRoutineLoadJob.modifyProperties`
converts the requested offsets before `modifyPropertiesInternal` installs the
new properties, so both the `getRealOffsets` and the `getOffsetsForTimes`
conversion went to the job's current group. It now passes the group this ALTER
declares, and keeps the current group only when the ALTER declares none.
`testCombinedAlterConvertsOffsetsInTheGroupItDeclares` alters a job bound to a
dropped group to a live one together with `kafka_offsets`, and checks that an
ALTER without `compute_group` still converts in the job's group.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
