github-actions[bot] commented on code in PR #69749: URL: https://github.com/apache/doris/pull/69749#discussion_r4408998664
########## 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: [P2] Give the exact-count Kafka test an isolated topic This topic survives the suite, while every invocation appends five more records and recreates an empty DUPLICATE KEY table. Because the job starts at OFFSET_BEGINNING, rerunning against the same broker with retained records (for example, a single-partition topic from the previous run) loads the previous five records plus the new five, so the exact assertEquals(5, ...) fails with ten even when compute-group binding works. Use a per-run topic with cleanup, or reset the fixed topic before producing, so the expected count covers only this invocation's records. ########## 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: [P2] Check the binding before initial Kafka metadata polling For a paused Kafka job without custom kafka_partitions, drop its declared group and RESUME it. RoutineLoadScheduler.process() first calls updateRoutineLoadJob(), which reaches RoutineLoadJob.update() -> refreshKafkaPartitions() -> KafkaUtil before any task is created. The absent group's empty backend list is converted into "no alive backends" and then a PARTITIONS_ERR with Kafka-property/topic troubleshooting text. The job is paused before reaching this check, and auto-resume repeats that misleading diagnostic. Run the declared-group check before the initial metadata lookup as well, while retaining this check for later tasks. The added missing-group unit test directly invokes scheduleOneTask with an existing task, so it bypasses this production path. -- 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]
