morningman commented on code in PR #69749:
URL: https://github.com/apache/doris/pull/69749#discussion_r4410635480
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java:
##########
@@ -135,9 +136,23 @@ private void scheduleOneTask(RoutineLoadTaskInfo
routineLoadTaskInfo) throws Exc
}
try {
- if
(routineLoadManager.getJob(routineLoadTaskInfo.getJobId()).isFinal()) {
+ RoutineLoadJob job =
routineLoadManager.getJob(routineLoadTaskInfo.getJobId());
+ if (job.isFinal()) {
return;
}
+
+ // The compute group the job declared can be dropped and the
owner's USAGE on it revoked
+ // while the job keeps running, so it is re-checked before every
task.
+ //
+ // This has to happen here rather than while the task is being
built. Backend allocation
+ // below resolves the backends through that same compute group, so
a missing group would
+ // otherwise pause the job with a generic "no available BE found"
before this check was
+ // ever reached, which is exactly the misleading message it exists
to replace. Running it
+ // first also keeps a failing check from leaving a transaction
behind, because beginTxn
+ // has not happened yet.
+
ComputeGroupBindingUtil.checkComputeGroupBeforeTask(job.getUserIdentity(),
Review Comment:
Fixed in 4bf82ff4a9d. `KafkaRoutineLoadJob.updateKafkaPartitions()` now runs
`ComputeGroupBindingUtil.checkComputeGroupBeforeTask` before the metadata
request. After RESUME, a job whose declared group is gone is paused with
`Compute group '<name>' not found.`, the pause the task scheduler gives,
instead of PARTITIONS_ERR with the Kafka advice. That pause keeps INTERNAL_ERR,
so auto resume picks the job up once the group is back, and a revoked USAGE
keeps CANNOT_RESUME_ERR. In other states the refresh only logs, as it does for
a failed metadata request. It returns instead of throwing, because
`RoutineLoadManager.updateRoutineLoadJob()` updates every job in one loop. The
new
`RoutineLoadComputeGroupTest#testDroppedComputeGroupPausesJobBeforeKafkaMetadataRequest`
drives `refreshKafkaPartitions` for a NEED_SCHEDULE job with a dropped group
and checks that no metadata request is sent and that the pause reason names the
group. It fails without the check.
##########
regression-test/suites/load_p0/routine_load/test_routine_load_compute_group.groovy:
##########
@@ -0,0 +1,290 @@
+// 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.kafka.clients.producer.KafkaProducer
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.clients.producer.ProducerConfig
+
+// Declaring `compute_group` on a routine load job.
+//
+// The declaration is stored in the job's property map under the key
`compute_group`, which is the
+// same key and value space the final (owner, compute_group, workload_group)
design uses, so the
+// metadata is read back as an explicit pin by later versions without
conversion.
+//
+// The property-level checks run before the Kafka data source is touched, so
the negative cases do
+// not need a broker; only the "job really runs there" cases do.
+suite("test_routine_load_compute_group", "p0") {
+ String tableName = "test_routine_load_compute_group_tbl"
+ String jobName = "test_routine_load_compute_group_job"
+
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName} (
+ `k1` INT NULL,
+ `k2` STRING NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`k1`)
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 1
+ PROPERTIES ("replication_allocation" = "tag.location.default: 1");
+ """
+
+ // Returns the statement instead of running it: inside a `test { }` block
the `sql` method must
+ // be the action's own, so the SQL has to be handed to it there rather
than executed here.
+ def createJobSql = { String computeGroup ->
+ return """
+ CREATE ROUTINE LOAD ${jobName} ON ${tableName}
+ COLUMNS TERMINATED BY ","
+ PROPERTIES ("compute_group" = "${computeGroup}")
+ FROM KAFKA ("kafka_broker_list" = "127.0.0.1:19092", "kafka_topic"
= "unused_topic");
+ """.toString()
+ }
+
+ if (!isCloudMode()) {
+ // Non-cloud is out of scope for this transitional change, so the key
can never appear in
+ // non-cloud metadata and upgrading such a cluster has nothing to
convert. The cloud-mode
+ // check runs first, so even the reserved DEFAULT is refused with this
message.
+ for (String value : ["any_group", "DEFAULT", "default"]) {
+ test {
+ sql createJobSql(value)
+ exception "only supported in cloud mode"
+ }
+ }
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+ return
+ }
+
+ // ---------------- cloud mode ----------------
+
+ // DEFAULT is reserved by the final design ("follow the owner's default
group"). A job pinned to
+ // a group literally named DEFAULT would be silently reinterpreted after
an upgrade.
+ for (String reserved : ["DEFAULT", "default", "Default"]) {
+ test {
+ sql createJobSql(reserved)
+ exception "reserved value"
+ }
+ }
+
+ test {
+ sql createJobSql("cg_that_does_not_exist")
+ exception "not found"
+ }
+
+ def currentComputeGroup = sql_return_maparray("show clusters")
+ .stream().filter(cg -> cg.is_current ==
"TRUE").findFirst().orElse(null)
+ assertNotNull(currentComputeGroup)
+ def cgName = currentComputeGroup.cluster
+ logger.info("current compute group: ${cgName}")
+
+ String enabled = context.config.otherConfigs.get("enableKafkaTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("kafka test not enabled, skipping the running-job part")
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+ return
+ }
+
+ String kafka_port = context.config.otherConfigs.get("kafka_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ def kafka_broker = "${externalEnvIp}:${kafka_port}"
+ def topic = "test_routine_load_compute_group"
Review Comment:
Fixed in 4bf82ff4a9d. The suite now produces to
`test_routine_load_compute_group_<uuid>`, a topic of its own on each run, as
`test_adaptive_random_bucket_routine_load` does, so the exact count only sees
this run's five records.
--
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]
