You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by he...@apache.org on 2026/04/16 02:55:40 UTC
(doris) branch master updated: [regression](routine_load_p0)add routine load docker cases (#62481)
This is an automated email from the ASF dual-hosted git repository.
hellostephen 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 7a7e70795de [regression](routine_load_p0)add routine load docker cases (#62481)
7a7e70795de is described below
commit 7a7e70795de3f9e7be1e84a2e1a5a8a608921888
Author: MoanasDaddyXu <xu...@selectdb.com>
AuthorDate: Thu Apr 16 10:55:32 2026 +0800
[regression](routine_load_p0)add routine load docker cases (#62481)
---
.../load/routineload/RoutineLoadTaskScheduler.java | 7 +
..._routine_load_master_switch_exactly_once.groovy | 194 +++++++++++++++++++
...mem_limit_exceeded_recovery_exactly_once.groovy | 192 ++++++++++++++++++
...mow_partial_update_concurrent_compaction.groovy | 196 +++++++++++++++++++
..._load_mow_partial_update_with_compaction.groovy | 211 ++++++++++++++++++++
...d_partial_update_new_key_error_atomicity.groovy | 214 ++++++++++++++++++++
...ne_load_partition_expansion_exactly_once.groovy | 156 +++++++++++++++
...ad_same_job_offset_recovery_exactly_once.groovy | 180 +++++++++++++++++
...load_same_job_schema_replay_exactly_once.groovy | 215 +++++++++++++++++++++
...oad_too_many_tasks_recovery_exactly_once.groovy | 177 +++++++++++++++++
10 files changed, 1742 insertions(+)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
index 4156410ef86..4de5f22c026 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
@@ -322,6 +322,13 @@ public class RoutineLoadTaskScheduler extends MasterDaemon {
throw new LoadException("MEM_LIMIT_EXCEEDED");
}
+ if (DebugPointUtil.isEnable("FE.ROUTINE_LOAD_TASK_SUBMIT_FAILED.TOO_MANY_TASKS")) {
+ LOG.warn("debug point FE.ROUTINE_LOAD_TASK_SUBMIT_FAILED.TOO_MANY_TASKS,"
+ + "routine load task submit failed");
+ tStatus = new TStatus(TStatusCode.TOO_MANY_TASKS);
+ tStatus.addToErrorMsgs("debug point: too many tasks");
+ }
+
if (tStatus.getStatusCode() != TStatusCode.OK) {
throw new LoadException("failed to submit task. error code: " + tStatus.getStatusCode()
+ ", msg: " + (tStatus.getErrorMsgsSize() > 0 ? tStatus.getErrorMsgs().get(0) : "NaN"));
diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load_master_switch_exactly_once.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load_master_switch_exactly_once.groovy
new file mode 100644
index 00000000000..94748cc650d
--- /dev/null
+++ b/regression-test/suites/load_p0/routine_load/test_routine_load_master_switch_exactly_once.groovy
@@ -0,0 +1,194 @@
+// 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.doris.regression.Config
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.RoutineLoadTestUtils
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.junit.Assert
+
+suite("test_routine_load_master_switch_exactly_once", "docker") {
+ def topicSuffix = System.currentTimeMillis()
+ def kafkaCsvTopic = "test_rl_master_switch_${topicSuffix}"
+
+ if (!RoutineLoadTestUtils.isKafkaTestEnabled(context)) {
+ return
+ }
+
+ def kafkaBroker = RoutineLoadTestUtils.getKafkaBroker(context)
+
+ def options = new ClusterOptions()
+ options.setFeNum(3)
+ options.setBeNum(1)
+ options.cloudMode = false
+
+ docker(options) {
+ def producer = RoutineLoadTestUtils.createKafkaProducer(kafkaBroker)
+
+ def tableName = "test_rl_master_switch_exactly_once"
+ def job = "test_rl_master_switch_job_${topicSuffix}"
+ Integer oldMasterIndex = null
+
+ def firstBatch = (1..30).collect { i ->
+ "${i},payload_${i},2026-04-${String.format('%02d', (i % 28) + 1)}".toString()
+ }
+ def secondBatch = (31..40).collect { i ->
+ "${i},payload_${i},2026-04-${String.format('%02d', (i % 28) + 1)}".toString()
+ }
+
+ def waitForRowCountAtLeast = { long expectedMinRows, int maxWait = 120 ->
+ def waited = 0
+ while (waited < maxWait) {
+ def rowCount = sql "SELECT count(*) FROM ${tableName}"
+ long count = (rowCount[0][0] as Number).longValue()
+ if (count >= expectedMinRows) {
+ return count
+ }
+ sleep(1000)
+ waited++
+ }
+ Assert.fail("Row count did not reach ${expectedMinRows} within timeout")
+ }
+
+ def waitForExactRowCount = { long expected, int maxWait = 180 ->
+ def waited = 0
+ while (waited < maxWait) {
+ def state = sql "SHOW ROUTINE LOAD FOR ${job}"
+ def rowCount = sql "SELECT count(*) FROM ${tableName}"
+ long count = (rowCount[0][0] as Number).longValue()
+ logger.info("waitForExactRowCount: state={}, rows={}", state[0][8].toString(), count)
+ if (state[0][8].toString() == "RUNNING" && count == expected) {
+ return
+ }
+ sleep(1000)
+ waited++
+ }
+ Assert.fail("Row count did not reach ${expected} within timeout")
+ }
+
+ def reconnectToCurrentMaster = {
+ def currentMaster = cluster.getMasterFe()
+ Assert.assertNotNull("Current master FE should exist", currentMaster)
+ def masterJdbcUrl = Config.buildUrlWithDb(currentMaster.host, currentMaster.queryPort, context.dbName)
+ context.connectTo(masterJdbcUrl, context.config.jdbcUser, context.config.jdbcPassword)
+ }
+
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName} (
+ `id` int NOT NULL,
+ `payload` string NULL,
+ `ts` string NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`id`)
+ DISTRIBUTED BY HASH(`id`) BUCKETS 1
+ PROPERTIES ("replication_allocation" = "tag.location.default: 1");
+ """
+ sql "sync"
+
+ try {
+ firstBatch.each { line ->
+ producer.send(new ProducerRecord<>(kafkaCsvTopic, null, line)).get()
+ }
+ producer.flush()
+
+ sql """
+ CREATE ROUTINE LOAD ${job} ON ${tableName}
+ COLUMNS TERMINATED BY ","
+ PROPERTIES
+ (
+ "max_batch_interval" = "5"
+ )
+ FROM KAFKA
+ (
+ "kafka_broker_list" = "${kafkaBroker}",
+ "kafka_topic" = "${kafkaCsvTopic}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING"
+ );
+ """
+
+ waitForRowCountAtLeast(1L)
+
+ def oldMaster = cluster.getMasterFe()
+ Assert.assertNotNull("Master FE should exist before switch", oldMaster)
+ oldMasterIndex = oldMaster.index
+ cluster.stopFrontends(oldMasterIndex)
+
+ def waited = 0
+ while (waited < 180) {
+ def newMaster = cluster.getMasterFe()
+ if (newMaster != null && newMaster.index != oldMasterIndex) {
+ logger.info("Master switched from {} to {}", oldMasterIndex, newMaster.index)
+ break
+ }
+ sleep(1000)
+ waited++
+ }
+ Assert.assertTrue("A new master FE should be elected after old master stops",
+ cluster.getMasterFe() != null && cluster.getMasterFe().index != oldMasterIndex)
+
+ reconnectToCurrentMaster()
+
+ secondBatch.each { line ->
+ producer.send(new ProducerRecord<>(kafkaCsvTopic, null, line)).get()
+ }
+ producer.flush()
+
+ waitForExactRowCount(40L)
+
+ def finalCount = sql "SELECT count(*), count(distinct id), sum(id) FROM ${tableName}"
+ Assert.assertEquals(40L, (finalCount[0][0] as Number).longValue())
+ Assert.assertEquals(40L, (finalCount[0][1] as Number).longValue())
+ Assert.assertEquals(820L, (finalCount[0][2] as Number).longValue())
+
+ def boundaryRows = sql "SELECT id, payload FROM ${tableName} WHERE id IN (1, 30, 31, 40) ORDER BY id"
+ Assert.assertEquals(4, boundaryRows.size())
+ Assert.assertEquals(1, boundaryRows[0][0])
+ Assert.assertEquals("payload_1", boundaryRows[0][1])
+ Assert.assertEquals(30, boundaryRows[1][0])
+ Assert.assertEquals("payload_30", boundaryRows[1][1])
+ Assert.assertEquals(31, boundaryRows[2][0])
+ Assert.assertEquals("payload_31", boundaryRows[2][1])
+ Assert.assertEquals(40, boundaryRows[3][0])
+ Assert.assertEquals("payload_40", boundaryRows[3][1])
+
+ def finalState = sql "SHOW ROUTINE LOAD FOR ${job}"
+ Assert.assertEquals("RUNNING", finalState[0][8].toString())
+ } finally {
+ if (oldMasterIndex != null) {
+ try {
+ cluster.startFrontends(oldMasterIndex)
+ sleep(5000)
+ } catch (Exception e) {
+ logger.warn("Failed to restart old master FE {}: {}", oldMasterIndex, e.message)
+ }
+ }
+ try {
+ reconnectToCurrentMaster()
+ } catch (Exception e) {
+ logger.warn("Failed to reconnect FE during cleanup: {}", e.message)
+ }
+ try {
+ sql "STOP ROUTINE LOAD FOR ${job}"
+ } catch (Exception e) {
+ logger.warn("Failed to stop routine load: {}", e.message)
+ }
+ producer.close()
+ sql "DROP TABLE IF EXISTS ${tableName}"
+ }
+ }
+}
diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load_mem_limit_exceeded_recovery_exactly_once.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load_mem_limit_exceeded_recovery_exactly_once.groovy
new file mode 100644
index 00000000000..b079a4db034
--- /dev/null
+++ b/regression-test/suites/load_p0/routine_load/test_routine_load_mem_limit_exceeded_recovery_exactly_once.groovy
@@ -0,0 +1,192 @@
+// 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.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.RoutineLoadTestUtils
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.junit.Assert
+
+suite("test_routine_load_mem_limit_exceeded_recovery_exactly_once", "docker") {
+ // 使用时间戳后缀确保每次运行使用不同 topic，避免历史消息污染精确计数断言
+ def topicSuffix = System.currentTimeMillis()
+ def kafkaCsvTopic = "test_rl_mem_limit_recovery_${topicSuffix}"
+
+ if (!RoutineLoadTestUtils.isKafkaTestEnabled(context)) {
+ return
+ }
+
+ def kafkaPort = context.config.otherConfigs.get("kafka_port")
+ Assert.assertNotNull("kafka_port must be configured in regression-conf.groovy", kafkaPort)
+ def kafka_broker = "127.0.0.1:${kafkaPort}"
+
+ def options = new ClusterOptions()
+ options.setFeNum(1)
+ options.setBeNum(1)
+ options.cloudMode = false
+ options.enableDebugPoints()
+
+ docker(options) {
+ def producer = RoutineLoadTestUtils.createKafkaProducer(kafka_broker)
+
+ def tableName = "test_rl_mem_limit_recovery"
+ def job = "test_mem_limit_recovery_job_${topicSuffix}"
+ def debugPoint = "FE.ROUTINE_LOAD_TASK_SUBMIT_FAILED.MEM_LIMIT_EXCEEDED"
+
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName} (
+ `id` int NOT NULL,
+ `payload` string NULL,
+ `ts` string NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`id`)
+ DISTRIBUTED BY HASH(`id`) BUCKETS 1
+ PROPERTIES ("replication_allocation" = "tag.location.default: 1");
+ """
+ sql "sync"
+
+ // 构造确定性测试数据：id=1..20
+ List<String> testData = []
+ for (int i = 1; i <= 20; i++) {
+ testData.add("${i},payload_${i},2026-01-${String.format('%02d', i)}".toString())
+ }
+
+ // 先发送数据到 Kafka
+ testData.each { line ->
+ logger.info("Sending to Kafka: ${line}")
+ def record = new ProducerRecord<>(kafkaCsvTopic, null, line.toString())
+ producer.send(record).get()
+ }
+ producer.flush()
+ logger.info("All 20 records sent to Kafka")
+
+ try {
+ // 第一阶段: 开启 debug point，模拟 MEM_LIMIT_EXCEEDED
+ // 该 debug point 在 RoutineLoadTaskScheduler.java (FE)，必须用 enableDebugPointForAllFEs
+ GetDebugPoint().enableDebugPointForAllFEs(debugPoint)
+ logger.info("Debug point enabled: ${debugPoint}")
+
+ sql """
+ CREATE ROUTINE LOAD ${job} ON ${tableName}
+ COLUMNS TERMINATED BY ","
+ PROPERTIES
+ (
+ "max_batch_interval" = "5"
+ )
+ FROM KAFKA
+ (
+ "kafka_broker_list" = "${kafka_broker}",
+ "kafka_topic" = "${kafkaCsvTopic}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING"
+ );
+ """
+
+ // 等待一段时间让任务提交失败
+ def waitCount = 0
+ def sawFailure = false
+ while (waitCount < 30) {
+ sleep(1000)
+ def state = sql "show routine load for ${job}"
+ def routineLoadState = state[0][8].toString()
+ def statistic = state[0][14].toString()
+ def otherMsg = state[0][19].toString()
+ logger.info("State: ${routineLoadState}, stats: ${statistic}, msg: ${otherMsg}")
+
+ // 解析 JSON 检查 abortedTaskNum 是否 > 0，这才是真实失败信号
+ // statistic.contains("abortedTaskNum") 始终为 true，不能作为判断依据
+ def statJson = new groovy.json.JsonSlurper().parseText(statistic)
+ long abortedCount = statJson.abortedTaskNum as long
+ if (abortedCount > 0 || otherMsg.contains("MEM_LIMIT_EXCEEDED")) {
+ sawFailure = true
+ logger.info("Detected MEM_LIMIT_EXCEEDED failure: abortedTaskNum=${abortedCount}, msg=${otherMsg}")
+ break
+ }
+ waitCount++
+ }
+ // 断言确实观察到了 debug point 生效引起的失败，防止 debug point 未命中时 case 悄悄通过
+ Assert.assertTrue("应观察到 MEM_LIMIT_EXCEEDED 失败（abortedTaskNum>0 或 otherMsg 含相关信息）", sawFailure)
+
+ // 检查失败阶段数据状态：应该没有完整消费
+ def failPhaseCount = sql "select count(*) from ${tableName}"
+ logger.info("Row count during failure phase: ${failPhaseCount[0][0]}")
+
+ // 第二阶段: 关闭 debug point，让 job 自动恢复
+ GetDebugPoint().disableDebugPointForAllFEs(debugPoint)
+ logger.info("Debug point disabled, waiting for recovery...")
+
+ // 等待 job 恢复并消费完成
+ def maxWaitCount = 120
+ def count = 0
+ while (true) {
+ def state = sql "show routine load for ${job}"
+ def routineLoadState = state[0][8].toString()
+ def statistic = state[0][14].toString()
+ logger.info("Recovery phase - State: ${routineLoadState}, stats: ${statistic}")
+
+ def rowCount = sql "select count(*) from ${tableName}"
+ if (routineLoadState == "RUNNING" && rowCount[0][0] == 20) {
+ logger.info("Job recovered and all 20 rows consumed")
+ break
+ }
+ if (count > maxWaitCount) {
+ Assert.fail("Routine load did not recover within timeout. " +
+ "State: ${routineLoadState}, rows: ${rowCount[0][0]}")
+ }
+ sleep(1000)
+ count++
+ }
+
+ // 核心断言: 精确不重不漏
+ def finalCount = sql "select count(*) from ${tableName}"
+ Assert.assertEquals("应精确消费 20 行", 20L, finalCount[0][0])
+
+ def distinctCount = sql "select count(distinct id) from ${tableName}"
+ Assert.assertEquals("不应有重复 id", 20L, distinctCount[0][0])
+
+ def sumId = sql "select sum(id) from ${tableName}"
+ Assert.assertEquals("sum(id) 应为 1+2+...+20=210", 210L, sumId[0][0])
+
+ def minId = sql "select min(id) from ${tableName}"
+ Assert.assertEquals("min(id) 应为 1", 1, minId[0][0])
+
+ def maxId = sql "select max(id) from ${tableName}"
+ Assert.assertEquals("max(id) 应为 20", 20, maxId[0][0])
+
+ // 验证 job 仍然是通过自然恢复的，而不是人工重建
+ def finalState = sql "show routine load for ${job}"
+ Assert.assertEquals("最终状态应为 RUNNING", "RUNNING", finalState[0][8].toString())
+ logger.info("All exactly-once assertions passed after MEM_LIMIT_EXCEEDED recovery")
+
+ } catch (Exception e) {
+ logger.error("Test failed: ${e.getMessage()}")
+ throw e
+ } finally {
+ try {
+ GetDebugPoint().disableDebugPointForAllFEs(debugPoint)
+ } catch (Exception e) {
+ logger.warn("Failed to disable debug point in cleanup: ${e.message}")
+ }
+ try {
+ sql "stop routine load for ${job}"
+ } catch (Exception e) {
+ logger.warn("Failed to stop routine load: ${e.message}")
+ }
+ producer.close()
+ sql "DROP TABLE IF EXISTS ${tableName}"
+ }
+ }
+}
diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load_mow_partial_update_concurrent_compaction.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load_mow_partial_update_concurrent_compaction.groovy
new file mode 100644
index 00000000000..e9d36528501
--- /dev/null
+++ b/regression-test/suites/load_p0/routine_load/test_routine_load_mow_partial_update_concurrent_compaction.groovy
@@ -0,0 +1,196 @@
+// 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 groovy.json.JsonSlurper
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.RoutineLoadTestUtils
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.junit.Assert
+
+suite("test_routine_load_mow_partial_update_concurrent_compaction", "docker") {
+ def topicSuffix = System.currentTimeMillis()
+ def kafkaCsvTopic = "test_rl_mow_pu_concurrent_compaction_${topicSuffix}"
+
+ if (!RoutineLoadTestUtils.isKafkaTestEnabled(context)) {
+ return
+ }
+
+ def kafkaBroker = RoutineLoadTestUtils.getKafkaBroker(context)
+
+ def options = new ClusterOptions()
+ options.setFeNum(1)
+ options.setBeNum(1)
+ options.cloudMode = false
+ options.enableDebugPoints()
+
+ docker(options) {
+ def producer = RoutineLoadTestUtils.createKafkaProducer(kafkaBroker)
+
+ def tableName = "test_rl_mow_pu_concurrent_compaction"
+ def job = "test_mow_pu_concurrent_compaction_job_${topicSuffix}"
+ def compactionDebugPoint = "FullCompaction.modify_rowsets.sleep"
+
+ def waitForLoadedRows = { long expectedLoadedRows, int maxWait = 120 ->
+ def waited = 0
+ while (waited < maxWait) {
+ def res = sql "SHOW ROUTINE LOAD FOR ${job}"
+ def state = res[0][8].toString()
+ def statJson = new JsonSlurper().parseText(res[0][14].toString())
+ long loaded = statJson.loadedRows as long
+ logger.info("waitForLoadedRows: state={}, loadedRows={}", state, loaded)
+ if (state == "RUNNING" && loaded >= expectedLoadedRows) {
+ return
+ }
+ sleep(1000)
+ waited++
+ }
+ Assert.fail("loadedRows did not reach ${expectedLoadedRows} within timeout")
+ }
+
+ sql """ DROP TABLE IF EXISTS ${tableName} force """
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName} (
+ `k` int NOT NULL,
+ `c1` int,
+ `c2` int,
+ `c3` int
+ ) ENGINE=OLAP
+ UNIQUE KEY(`k`)
+ DISTRIBUTED BY HASH(`k`) BUCKETS 1
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1",
+ "enable_unique_key_merge_on_write" = "true",
+ "disable_auto_compaction" = "true"
+ );
+ """
+
+ sql """
+ INSERT INTO ${tableName} VALUES
+ (1, 1, 11, 111),
+ (2, 2, 22, 222),
+ (3, 3, 33, 333)
+ """
+ sql "sync"
+
+ def batch1 = [
+ "1,10,110",
+ "2,20,220"
+ ]
+ def batch2 = [
+ "1,11,1110",
+ "3,30,330"
+ ]
+ def batch3 = [
+ "2,21,2210",
+ "3,31,3310"
+ ]
+
+ def compactionError = null
+ Thread compactionThread = null
+
+ try {
+ sql """
+ CREATE ROUTINE LOAD ${job} ON ${tableName}
+ COLUMNS TERMINATED BY ",",
+ COLUMNS (k, c1, c2)
+ PROPERTIES
+ (
+ "max_batch_interval" = "1",
+ "partial_columns" = "true"
+ )
+ FROM KAFKA
+ (
+ "kafka_broker_list" = "${kafkaBroker}",
+ "kafka_topic" = "${kafkaCsvTopic}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING"
+ );
+ """
+
+ batch1.each { line ->
+ producer.send(new ProducerRecord<>(kafkaCsvTopic, null, line)).get()
+ }
+ producer.flush()
+ waitForLoadedRows(2L)
+
+ GetDebugPoint().enableDebugPointForAllBEs(compactionDebugPoint)
+ compactionThread = Thread.start {
+ try {
+ trigger_and_wait_compaction(tableName, "full")
+ } catch (Throwable t) {
+ compactionError = t
+ }
+ }
+
+ sleep(2000)
+
+ batch2.each { line ->
+ producer.send(new ProducerRecord<>(kafkaCsvTopic, null, line)).get()
+ }
+ producer.flush()
+
+ sleep(1500)
+
+ batch3.each { line ->
+ producer.send(new ProducerRecord<>(kafkaCsvTopic, null, line)).get()
+ }
+ producer.flush()
+
+ waitForLoadedRows(6L, 180)
+
+ compactionThread.join(120000)
+ Assert.assertFalse("Compaction thread should finish within timeout", compactionThread.isAlive())
+ if (compactionError != null) {
+ throw compactionError
+ }
+
+ def finalRows = sql "SELECT k, c1, c2, c3 FROM ${tableName} ORDER BY k"
+ Assert.assertEquals(3, finalRows.size())
+ Assert.assertEquals(1, finalRows[0][0])
+ Assert.assertEquals(11, finalRows[0][1])
+ Assert.assertEquals(1110, finalRows[0][2])
+ Assert.assertEquals(111, finalRows[0][3])
+
+ Assert.assertEquals(2, finalRows[1][0])
+ Assert.assertEquals(21, finalRows[1][1])
+ Assert.assertEquals(2210, finalRows[1][2])
+ Assert.assertEquals(222, finalRows[1][3])
+
+ Assert.assertEquals(3, finalRows[2][0])
+ Assert.assertEquals(31, finalRows[2][1])
+ Assert.assertEquals(3310, finalRows[2][2])
+ Assert.assertEquals(333, finalRows[2][3])
+
+ def finalCount = sql "SELECT count(*), count(distinct k), sum(k) FROM ${tableName}"
+ Assert.assertEquals(3L, (finalCount[0][0] as Number).longValue())
+ Assert.assertEquals(3L, (finalCount[0][1] as Number).longValue())
+ Assert.assertEquals(6L, (finalCount[0][2] as Number).longValue())
+ } finally {
+ try {
+ GetDebugPoint().disableDebugPointForAllBEs(compactionDebugPoint)
+ } catch (Exception e) {
+ logger.warn("Failed to disable compaction debug point: {}", e.message)
+ }
+ try {
+ sql "STOP ROUTINE LOAD FOR ${job}"
+ } catch (Exception e) {
+ logger.warn("Failed to stop routine load: {}", e.message)
+ }
+ producer.close()
+ sql "DROP TABLE IF EXISTS ${tableName} force"
+ }
+ }
+}
diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load_mow_partial_update_with_compaction.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load_mow_partial_update_with_compaction.groovy
new file mode 100644
index 00000000000..f88a8eae8b8
--- /dev/null
+++ b/regression-test/suites/load_p0/routine_load/test_routine_load_mow_partial_update_with_compaction.groovy
@@ -0,0 +1,211 @@
+// 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.doris.regression.util.RoutineLoadTestUtils
+import org.apache.kafka.clients.producer.KafkaProducer
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.junit.Assert
+
+suite("test_routine_load_mow_partial_update_with_compaction", "nonConcurrent") {
+ // 使用时间戳后缀确保每次运行使用不同 topic，避免历史消息污染精确计数断言
+ def topicSuffix = System.currentTimeMillis()
+ def kafkaCsvTopic = "test_rl_mow_pu_compaction_${topicSuffix}"
+
+ if (RoutineLoadTestUtils.isKafkaTestEnabled(context)) {
+ def runSql = { String q -> sql q }
+ def kafka_broker = RoutineLoadTestUtils.getKafkaBroker(context)
+ def producer = RoutineLoadTestUtils.createKafkaProducer(kafka_broker)
+
+ def tableName = "test_rl_mow_pu_compaction"
+ def job = "test_mow_pu_compaction_job_${topicSuffix}"
+
+ // 基于 loadedRows 统计的等待 helper，适用于 MOW partial update（行数不增加的场景）
+ def waitForLoadedRows = { String jobName, long expectedLoadedRows, int maxWait = 60 ->
+ def waited = 0
+ while (waited < maxWait) {
+ def res = sql "show routine load for ${jobName}"
+ def state = res[0][8].toString()
+ def statJson = new groovy.json.JsonSlurper().parseText(res[0][14].toString())
+ long loaded = statJson.loadedRows as long
+ logger.info("waitForLoadedRows: state=${state}, loadedRows=${loaded}, expected>=${expectedLoadedRows}")
+ if (state == "RUNNING" && loaded >= expectedLoadedRows) break
+ if (waited >= maxWait - 1) {
+ Assert.fail("Timeout waiting for loadedRows >= ${expectedLoadedRows}, got ${loaded}")
+ }
+ sleep(1000)
+ waited++
+ }
+ }
+
+ sql """ DROP TABLE IF EXISTS ${tableName} force;"""
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName} (
+ `k` int NOT NULL,
+ `c1` int,
+ `c2` int,
+ `c3` int
+ ) ENGINE=OLAP
+ UNIQUE KEY(`k`)
+ DISTRIBUTED BY HASH(`k`) BUCKETS 1
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1",
+ "enable_unique_key_merge_on_write" = "true",
+ "disable_auto_compaction" = "true"
+ );
+ """
+
+ // 插入基线数据
+ sql """
+ INSERT INTO ${tableName} VALUES
+ (1, 1, 11, 111),
+ (2, 2, 22, 222),
+ (3, 3, 33, 333)
+ """
+ sql "sync"
+
+ // 验证基线
+ def baselineRows = sql "SELECT k, c1, c2, c3 FROM ${tableName} ORDER BY k"
+ Assert.assertEquals(3, baselineRows.size())
+ logger.info("Baseline data: ${baselineRows}")
+
+ try {
+ sql """
+ CREATE ROUTINE LOAD ${job} ON ${tableName}
+ COLUMNS TERMINATED BY ",",
+ COLUMNS (k, c1, c2)
+ PROPERTIES
+ (
+ "max_batch_interval" = "10",
+ "partial_columns" = "true"
+ )
+ FROM KAFKA
+ (
+ "kafka_broker_list" = "${kafka_broker}",
+ "kafka_topic" = "${kafkaCsvTopic}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING"
+ );
+ """
+
+ // 第一批 partial update: 更新 key=1,2 的 c1,c2
+ def batch1 = [
+ "1,10,110",
+ "2,20,220"
+ ]
+ batch1.each { line ->
+ logger.info("Sending batch1: ${line}")
+ def record = new ProducerRecord<>(kafkaCsvTopic, null, line)
+ producer.send(record).get()
+ }
+ producer.flush()
+
+ // 等待第一批消费完成（batch1 共 2 条，累计 loadedRows >= 2）
+ // 不用 waitForTaskFinish(count>3)，因为基线已有 3 行，行数不会增加
+ waitForLoadedRows(job, 2)
+ logger.info("Batch1 consumed, triggering full compaction...")
+
+ // 暂停 routine load，避免 compaction 和消费并行引发不确定性
+ sql "PAUSE ROUTINE LOAD FOR ${job}"
+
+ // 等待 job 进入 PAUSED
+ def pauseWait = 0
+ while (pauseWait < 30) {
+ def state = sql "show routine load for ${job}"
+ if (state[0][8].toString() == "PAUSED") break
+ sleep(1000)
+ pauseWait++
+ }
+ def pausedState = sql "show routine load for ${job}"
+ Assert.assertEquals("Routine load 应成功进入 PAUSED 再触发 compaction", "PAUSED", pausedState[0][8].toString())
+
+ // 触发 full compaction
+ trigger_and_wait_compaction(tableName, "full")
+ logger.info("Full compaction completed after batch1")
+
+ // 恢复 routine load
+ sql "RESUME ROUTINE LOAD FOR ${job}"
+
+ // 第二批 partial update: 更新 key=1,3 的 c1,c2
+ def batch2 = [
+ "1,11,1110",
+ "3,30,330"
+ ]
+ batch2.each { line ->
+ logger.info("Sending batch2: ${line}")
+ def record = new ProducerRecord<>(kafkaCsvTopic, null, line)
+ producer.send(record).get()
+ }
+ producer.flush()
+
+ // 等待第二批消费完成（batch1+batch2 累计 loadedRows >= 4）
+ // 同上不用行数判断
+ waitForLoadedRows(job, 4)
+
+ // 核心断言 1: 最终行数
+ def finalCount = sql "SELECT count(*) FROM ${tableName}"
+ Assert.assertEquals("应有 3 行", 3L, finalCount[0][0])
+
+ def distinctCount = sql "SELECT count(distinct k) FROM ${tableName}"
+ Assert.assertEquals("不应有重复 key", 3L, distinctCount[0][0])
+
+ // 核心断言 2: 按主键排序后的精确行集
+ // key=1: c1 被第二批覆盖为 11, c2 被覆盖为 1110, c3 不变 111
+ // key=2: c1 被第一批更新为 20, c2 更新为 220, c3 不变 222
+ // key=3: c1 被第二批更新为 30, c2 更新为 330, c3 不变 333
+ def finalRows = sql "SELECT k, c1, c2, c3 FROM ${tableName} ORDER BY k"
+ logger.info("Final rows: ${finalRows}")
+
+ Assert.assertEquals(3, finalRows.size())
+
+ // key=1
+ Assert.assertEquals(1, finalRows[0][0])
+ Assert.assertEquals(11, finalRows[0][1])
+ Assert.assertEquals(1110, finalRows[0][2])
+ Assert.assertEquals(111, finalRows[0][3])
+
+ // key=2
+ Assert.assertEquals(2, finalRows[1][0])
+ Assert.assertEquals(20, finalRows[1][1])
+ Assert.assertEquals(220, finalRows[1][2])
+ Assert.assertEquals(222, finalRows[1][3])
+
+ // key=3
+ Assert.assertEquals(3, finalRows[2][0])
+ Assert.assertEquals(30, finalRows[2][1])
+ Assert.assertEquals(330, finalRows[2][2])
+ Assert.assertEquals(333, finalRows[2][3])
+
+ // 核心断言 3: 聚合校验
+ def aggResult = sql "SELECT count(*), sum(k) FROM ${tableName}"
+ Assert.assertEquals(3L, aggResult[0][0])
+ Assert.assertEquals(6L, aggResult[0][1])
+
+ logger.info("All assertions passed: MOW partial update + compaction correctness verified")
+
+ } catch (Exception e) {
+ logger.error("Test failed: ${e.getMessage()}")
+ throw e
+ } finally {
+ try {
+ sql "STOP ROUTINE LOAD FOR ${job}"
+ } catch (Exception e) {
+ logger.warn("Failed to stop routine load: ${e.message}")
+ }
+ producer.close()
+ sql "DROP TABLE IF EXISTS ${tableName} force"
+ }
+ }
+}
diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load_partial_update_new_key_error_atomicity.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load_partial_update_new_key_error_atomicity.groovy
new file mode 100644
index 00000000000..060c6fd0563
--- /dev/null
+++ b/regression-test/suites/load_p0/routine_load/test_routine_load_partial_update_new_key_error_atomicity.groovy
@@ -0,0 +1,214 @@
+// 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.doris.regression.util.RoutineLoadTestUtils
+import org.apache.kafka.clients.producer.KafkaProducer
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.junit.Assert
+
+suite("test_routine_load_partial_update_new_key_error_atomicity", "nonConcurrent") {
+ // 使用时间戳后缀确保每次运行使用不同 topic，避免历史消息污染
+ def topicSuffix = System.currentTimeMillis()
+ // Phase-1: 原子性测试 topic（含新 key 的坏批次）
+ def kafkaBadTopic = "test_rl_pu_new_key_error_bad_${topicSuffix}"
+ // Phase-2: 恢复测试 topic（仅老 key 的合法数据）
+ def kafkaGoodTopic = "test_rl_pu_new_key_error_good_${topicSuffix}"
+
+ if (RoutineLoadTestUtils.isKafkaTestEnabled(context)) {
+ def runSql = { String q -> sql q }
+ def kafka_broker = RoutineLoadTestUtils.getKafkaBroker(context)
+ def producer = RoutineLoadTestUtils.createKafkaProducer(kafka_broker)
+
+ def tableName = "test_rl_pu_new_key_error_atomicity"
+ def badJob = "test_pu_new_key_error_atomicity_job_${topicSuffix}"
+ def recoveryJob = "test_pu_new_key_error_recovery_job_${topicSuffix}"
+
+ // 基于 loadedRows 的等待 helper
+ def waitForLoadedRows = { String jobName, long expectedLoadedRows, int maxWait = 60 ->
+ def waited = 0
+ while (waited < maxWait) {
+ def res = sql "show routine load for ${jobName}"
+ def state = res[0][8].toString()
+ def statJson = new groovy.json.JsonSlurper().parseText(res[0][14].toString())
+ long loaded = statJson.loadedRows as long
+ logger.info("waitForLoadedRows[${jobName}]: state=${state}, loadedRows=${loaded}, expected>=${expectedLoadedRows}")
+ if (state == "RUNNING" && loaded >= expectedLoadedRows) break
+ if (waited >= maxWait - 1) {
+ Assert.fail("Timeout waiting for loadedRows >= ${expectedLoadedRows}, got ${loaded}")
+ }
+ sleep(1000)
+ waited++
+ }
+ }
+
+ sql """ DROP TABLE IF EXISTS ${tableName} force;"""
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName} (
+ `k` int NOT NULL,
+ `c1` int,
+ `c2` int
+ ) ENGINE=OLAP
+ UNIQUE KEY(`k`)
+ DISTRIBUTED BY HASH(`k`) BUCKETS 1
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1",
+ "enable_unique_key_merge_on_write" = "true"
+ );
+ """
+
+ // 插入基线数据
+ sql """
+ INSERT INTO ${tableName} VALUES
+ (1, 1, 1),
+ (2, 2, 2),
+ (3, 3, 3)
+ """
+ sql "sync"
+
+ // 验证基线数据
+ def baselineRows = sql "SELECT * FROM ${tableName} ORDER BY k"
+ Assert.assertEquals(3, baselineRows.size())
+ logger.info("Baseline data verified: ${baselineRows}")
+
+ try {
+ sql """
+ CREATE ROUTINE LOAD ${badJob} ON ${tableName}
+ COLUMNS TERMINATED BY ",",
+ COLUMNS (k, c1)
+ PROPERTIES
+ (
+ "max_batch_interval" = "10",
+ "partial_columns" = "true",
+ "partial_update_new_key_behavior" = "error"
+ )
+ FROM KAFKA
+ (
+ "kafka_broker_list" = "${kafka_broker}",
+ "kafka_topic" = "${kafkaBadTopic}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING"
+ );
+ """
+
+ // 发送包含老 key 和新 key 的混合数据
+ // 在 new_key_behavior=error 模式下，新 key 应导致整批失败
+ def mixedData = [
+ "1,100", // 老 key — 合法的 partial update
+ "2,200", // 老 key — 合法的 partial update
+ "10,1000", // 新 key — 非法，应触发 error
+ "11,1100" // 新 key — 非法，应触发 error
+ ]
+
+ mixedData.each { line ->
+ logger.info("Sending mixed data to Kafka bad topic: ${line}")
+ def record = new ProducerRecord<>(kafkaBadTopic, null, line)
+ producer.send(record).get()
+ }
+ producer.flush()
+
+ // 等待任务失败（task abort）
+ RoutineLoadTestUtils.waitForTaskAbort(runSql, badJob)
+ def state = sql "show routine load for ${badJob}"
+ logger.info("Job state after mixed batch: ${state[0][8].toString()}")
+ logger.info("Job error info: ${state[0][17].toString()}")
+
+ // 核心断言 1: 坏批次整体失败，老 key 不应被局部更新
+ // 表数据应仍为基线值
+ def afterFailRows = sql "SELECT * FROM ${tableName} ORDER BY k"
+ Assert.assertEquals("基线数据行数应不变", 3, afterFailRows.size())
+ Assert.assertEquals("key=1 的 c1 应仍为基线值 1", 1, afterFailRows[0][1])
+ Assert.assertEquals("key=2 的 c1 应仍为基线值 2", 2, afterFailRows[1][1])
+ Assert.assertEquals("key=3 的 c1 应仍为基线值 3", 3, afterFailRows[2][1])
+ // 不应有新 key 10, 11
+ def newKeyCount = sql "SELECT count(*) FROM ${tableName} WHERE k IN (10, 11)"
+ Assert.assertEquals("不应存在新 key", 0L, newKeyCount[0][0])
+ logger.info("Atomicity assertion passed: mixed batch fully rejected")
+
+ // Phase-2: 恢复阶段
+ // 原 job (badJob) 会不断重试同一坏 offset，不能直接 resume。
+ // 方案：停掉 badJob，用干净的 recoveryTopic 创建新 job 验证恢复能力。
+ sql "STOP ROUTINE LOAD FOR ${badJob}"
+
+ // 发送只包含老 key 的合法数据到恢复 topic
+ def goodData = [
+ "1,101",
+ "2,202"
+ ]
+ goodData.each { line ->
+ logger.info("Sending good data to recovery topic: ${line}")
+ def record = new ProducerRecord<>(kafkaGoodTopic, null, line)
+ producer.send(record).get()
+ }
+ producer.flush()
+
+ // 创建新 job，指向干净的 recoveryTopic
+ sql """
+ CREATE ROUTINE LOAD ${recoveryJob} ON ${tableName}
+ COLUMNS TERMINATED BY ",",
+ COLUMNS (k, c1)
+ PROPERTIES
+ (
+ "max_batch_interval" = "5",
+ "partial_columns" = "true",
+ "partial_update_new_key_behavior" = "error"
+ )
+ FROM KAFKA
+ (
+ "kafka_broker_list" = "${kafka_broker}",
+ "kafka_topic" = "${kafkaGoodTopic}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING"
+ );
+ """
+
+ // 等待 recovery job 消费 2 条合法数据（MOW partial update 行数不增加，用 loadedRows）
+ waitForLoadedRows(recoveryJob, 2)
+
+ // 核心断言 2: 恢复 job 成功写入合法数据。
+ // 这里检查的是用户可见的最终结果，不应开启 skip_delete_bitmap。
+ def finalRows = sql "SELECT k, c1, c2 FROM ${tableName} ORDER BY k"
+ logger.info("Final rows after recovery: ${finalRows}")
+ Assert.assertEquals("最终应有 3 行", 3, finalRows.size())
+ // key=1: c1 应更新为 101，c2 保持 1
+ Assert.assertEquals(1, finalRows[0][0])
+ Assert.assertEquals(101, finalRows[0][1])
+ Assert.assertEquals(1, finalRows[0][2])
+ // key=2: c1 应更新为 202，c2 保持 2
+ Assert.assertEquals(2, finalRows[1][0])
+ Assert.assertEquals(202, finalRows[1][1])
+ Assert.assertEquals(2, finalRows[1][2])
+ // key=3: 不变
+ Assert.assertEquals(3, finalRows[2][0])
+ Assert.assertEquals(3, finalRows[2][1])
+ Assert.assertEquals(3, finalRows[2][2])
+
+ // 聚合断言
+ def aggResult = sql "SELECT count(*), count(distinct k), sum(k) FROM ${tableName}"
+ Assert.assertEquals(3L, aggResult[0][0])
+ Assert.assertEquals(3L, aggResult[0][1])
+ Assert.assertEquals(6L, aggResult[0][2])
+ logger.info("Recovery assertion passed: good batch consumed correctly via recovery job")
+
+ } catch (Exception e) {
+ logger.error("Test failed: ${e.getMessage()}")
+ throw e
+ } finally {
+ try { sql "STOP ROUTINE LOAD FOR ${badJob}" } catch (Exception e) { logger.warn("stop badJob: ${e.message}") }
+ try { sql "STOP ROUTINE LOAD FOR ${recoveryJob}" } catch (Exception e) { logger.warn("stop recoveryJob: ${e.message}") }
+ producer.close()
+ sql "DROP TABLE IF EXISTS ${tableName} force"
+ }
+ }
+}
diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load_partition_expansion_exactly_once.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load_partition_expansion_exactly_once.groovy
new file mode 100644
index 00000000000..90db3bd0039
--- /dev/null
+++ b/regression-test/suites/load_p0/routine_load/test_routine_load_partition_expansion_exactly_once.groovy
@@ -0,0 +1,156 @@
+// 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.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.RoutineLoadTestUtils
+import org.apache.kafka.clients.admin.AdminClient
+import org.apache.kafka.clients.admin.NewPartitions
+import org.apache.kafka.clients.admin.NewTopic
+import org.apache.kafka.clients.producer.ProducerConfig
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.junit.Assert
+import java.util.Collections
+
+suite("test_routine_load_partition_expansion_exactly_once", "docker") {
+ def topicSuffix = System.currentTimeMillis()
+ String kafkaCsvTopic = "test_rl_partition_expansion_${topicSuffix}"
+
+ if (!RoutineLoadTestUtils.isKafkaTestEnabled(context)) {
+ return
+ }
+
+ def kafkaBroker = RoutineLoadTestUtils.getKafkaBroker(context)
+
+ def options = new ClusterOptions()
+ options.setFeNum(1)
+ options.setBeNum(1)
+ options.cloudMode = false
+
+ docker(options) {
+ def producer = RoutineLoadTestUtils.createKafkaProducer(kafkaBroker)
+
+ def adminProps = new Properties()
+ adminProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBroker)
+ def adminClient = AdminClient.create(adminProps)
+
+ def tableName = "test_rl_partition_expansion"
+ def job = "test_rl_partition_expansion_job_${topicSuffix}"
+
+ def waitForExactRowCount = { long expected, int maxWait = 180 ->
+ def waited = 0
+ while (waited < maxWait) {
+ def state = sql "SHOW ROUTINE LOAD FOR ${job}"
+ def rowCount = sql "SELECT count(*) FROM ${tableName}"
+ long count = (rowCount[0][0] as Number).longValue()
+ logger.info("waitForExactRowCount: state={}, rows={}", state[0][8].toString(), count)
+ if (state[0][8].toString() == "RUNNING" && count == expected) {
+ return
+ }
+ sleep(1000)
+ waited++
+ }
+ Assert.fail("Row count did not reach ${expected} within timeout")
+ }
+
+ def waitForPartitionCount = { int expected, int maxWait = 60 ->
+ def waited = 0
+ while (waited < maxWait) {
+ def topicDesc = adminClient.describeTopics([kafkaCsvTopic]).values().get(kafkaCsvTopic).get()
+ if (topicDesc.partitions().size() == expected) {
+ return
+ }
+ sleep(1000)
+ waited++
+ }
+ Assert.fail("Kafka topic partition count did not reach ${expected}")
+ }
+
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName} (
+ `id` int NOT NULL,
+ `payload` string NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`id`)
+ DISTRIBUTED BY HASH(`id`) BUCKETS 1
+ PROPERTIES ("replication_allocation" = "tag.location.default: 1");
+ """
+ sql "sync"
+
+ try {
+ adminClient.createTopics([new NewTopic(kafkaCsvTopic, 1, (short) 1)]).all().get()
+ waitForPartitionCount(1)
+
+ sql """
+ CREATE ROUTINE LOAD ${job} ON ${tableName}
+ COLUMNS TERMINATED BY ","
+ PROPERTIES
+ (
+ "max_batch_interval" = "5"
+ )
+ FROM KAFKA
+ (
+ "kafka_broker_list" = "${kafkaBroker}",
+ "kafka_topic" = "${kafkaCsvTopic}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING"
+ );
+ """
+
+ (1..4).each { i ->
+ producer.send(new ProducerRecord<>(kafkaCsvTopic, 0, null, "${i},payload_${i}".toString())).get()
+ }
+ producer.flush()
+ waitForExactRowCount(4L)
+
+ adminClient.createPartitions(Collections.singletonMap(kafkaCsvTopic, NewPartitions.increaseTo(3))).all().get()
+ waitForPartitionCount(3)
+
+ producer.send(new ProducerRecord<>(kafkaCsvTopic, 1, null, "5,payload_5")).get()
+ producer.send(new ProducerRecord<>(kafkaCsvTopic, 2, null, "6,payload_6")).get()
+ producer.send(new ProducerRecord<>(kafkaCsvTopic, 1, null, "7,payload_7")).get()
+ producer.send(new ProducerRecord<>(kafkaCsvTopic, 2, null, "8,payload_8")).get()
+ producer.send(new ProducerRecord<>(kafkaCsvTopic, 0, null, "9,payload_9")).get()
+ producer.flush()
+
+ waitForExactRowCount(9L, 240)
+
+ def finalRows = sql "SELECT id, payload FROM ${tableName} ORDER BY id"
+ Assert.assertEquals(9, finalRows.size())
+ for (int i = 0; i < 9; i++) {
+ Assert.assertEquals(i + 1, finalRows[i][0])
+ Assert.assertEquals("payload_${i + 1}".toString(), finalRows[i][1])
+ }
+
+ def aggregate = sql "SELECT count(*), count(distinct id), sum(id) FROM ${tableName}"
+ Assert.assertEquals(9L, (aggregate[0][0] as Number).longValue())
+ Assert.assertEquals(9L, (aggregate[0][1] as Number).longValue())
+ Assert.assertEquals(45L, (aggregate[0][2] as Number).longValue())
+
+ def finalState = sql "SHOW ROUTINE LOAD FOR ${job}"
+ Assert.assertEquals("RUNNING", finalState[0][8].toString())
+ } finally {
+ try {
+ sql "STOP ROUTINE LOAD FOR ${job}"
+ } catch (Exception e) {
+ logger.warn("Failed to stop routine load: {}", e.message)
+ }
+ producer.close()
+ adminClient.close()
+ sql "DROP TABLE IF EXISTS ${tableName}"
+ }
+ }
+}
diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load_same_job_offset_recovery_exactly_once.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load_same_job_offset_recovery_exactly_once.groovy
new file mode 100644
index 00000000000..2f6cd068696
--- /dev/null
+++ b/regression-test/suites/load_p0/routine_load/test_routine_load_same_job_offset_recovery_exactly_once.groovy
@@ -0,0 +1,180 @@
+// 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.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.RoutineLoadTestUtils
+import org.apache.kafka.clients.admin.AdminClient
+import org.apache.kafka.clients.admin.NewTopic
+import org.apache.kafka.clients.producer.ProducerConfig
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.junit.Assert
+
+suite("test_routine_load_same_job_offset_recovery_exactly_once", "docker") {
+ def topicSuffix = System.currentTimeMillis()
+ def kafkaCsvTopic = "test_rl_same_job_offset_recovery_${topicSuffix}"
+
+ if (!RoutineLoadTestUtils.isKafkaTestEnabled(context)) {
+ return
+ }
+
+ def kafkaBroker = RoutineLoadTestUtils.getKafkaBroker(context)
+
+ def options = new ClusterOptions()
+ options.setFeNum(1)
+ options.setBeNum(1)
+ options.cloudMode = false
+
+ docker(options) {
+ def producer = RoutineLoadTestUtils.createKafkaProducer(kafkaBroker)
+
+ def adminProps = new Properties()
+ adminProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBroker)
+ def adminClient = AdminClient.create(adminProps)
+
+ def tableName = "test_rl_same_job_offset_recovery"
+ def job = "test_rl_same_job_offset_recovery_job_${topicSuffix}"
+
+ def firstBatch = [
+ "1,payload_1,2026-03-01",
+ "2,payload_2,2026-03-02"
+ ]
+ def secondBatch = [
+ "3,payload_3,2026-03-03",
+ "4,payload_4,2026-03-04",
+ "5,payload_5,2026-03-05"
+ ]
+
+ def waitForExactRowCount = { long expected, int maxWait = 120 ->
+ def waited = 0
+ while (waited < maxWait) {
+ def state = sql "SHOW ROUTINE LOAD FOR ${job}"
+ def rowCount = sql "SELECT count(*) FROM ${tableName}"
+ long count = (rowCount[0][0] as Number).longValue()
+ logger.info("waitForExactRowCount: state={}, rows={}", state[0][8].toString(), count)
+ if (state[0][8].toString() == "RUNNING" && count == expected) {
+ return
+ }
+ sleep(1000)
+ waited++
+ }
+ Assert.fail("Row count did not reach ${expected} within timeout")
+ }
+
+ def waitForPausedByOffsetError = { int maxWait = 120 ->
+ def waited = 0
+ while (waited < maxWait) {
+ def res = sql "SHOW ROUTINE LOAD FOR ${job}"
+ def state = res[0][8].toString()
+ def reason = res[0][17].toString()
+ def otherMsg = res[0][19].toString()
+ def combinedMsg = "${reason} ${otherMsg}".toLowerCase()
+ logger.info("waitForPausedByOffsetError: state={}, reason={}, msg={}", state, reason, otherMsg)
+ if (state == "PAUSED" && (
+ combinedMsg.contains("offset out of range")
+ || combinedMsg.contains("greater than kafka latest offset")
+ )) {
+ return
+ }
+ sleep(1000)
+ waited++
+ }
+ Assert.fail("Routine load job did not pause on bad offset within timeout")
+ }
+
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName} (
+ `id` int NOT NULL,
+ `payload` string NULL,
+ `ts` string NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`id`)
+ DISTRIBUTED BY HASH(`id`) BUCKETS 1
+ PROPERTIES ("replication_allocation" = "tag.location.default: 1");
+ """
+ sql "sync"
+
+ try {
+ adminClient.createTopics([new NewTopic(kafkaCsvTopic, 1, (short) 1)]).all().get()
+
+ sql """
+ CREATE ROUTINE LOAD ${job} ON ${tableName}
+ COLUMNS TERMINATED BY ","
+ PROPERTIES
+ (
+ "max_batch_interval" = "5"
+ )
+ FROM KAFKA
+ (
+ "kafka_broker_list" = "${kafkaBroker}",
+ "kafka_topic" = "${kafkaCsvTopic}",
+ "kafka_partitions" = "0",
+ "kafka_offsets" = "OFFSET_BEGINNING"
+ );
+ """
+
+ firstBatch.each { line ->
+ producer.send(new ProducerRecord<>(kafkaCsvTopic, 0, null, line)).get()
+ }
+ producer.flush()
+ waitForExactRowCount(2L)
+
+ sql "PAUSE ROUTINE LOAD FOR ${job}"
+ sql "ALTER ROUTINE LOAD FOR ${job} FROM KAFKA(\"kafka_partitions\" = \"0\", \"kafka_offsets\" = \"999999\");"
+ sql "RESUME ROUTINE LOAD FOR ${job}"
+
+ waitForPausedByOffsetError()
+
+ def pausedCount = sql "SELECT count(*) FROM ${tableName}"
+ Assert.assertEquals(2L, (pausedCount[0][0] as Number).longValue())
+
+ sql "ALTER ROUTINE LOAD FOR ${job} FROM KAFKA(\"kafka_partitions\" = \"0\", \"kafka_offsets\" = \"2\");"
+ secondBatch.each { line ->
+ producer.send(new ProducerRecord<>(kafkaCsvTopic, 0, null, line)).get()
+ }
+ producer.flush()
+ sql "RESUME ROUTINE LOAD FOR ${job}"
+
+ waitForExactRowCount(5L)
+
+ def finalRows = sql "SELECT id, payload FROM ${tableName} ORDER BY id"
+ Assert.assertEquals(5, finalRows.size())
+ for (int i = 0; i < 5; i++) {
+ Assert.assertEquals(i + 1, finalRows[i][0])
+ Assert.assertEquals("payload_${i + 1}".toString(), finalRows[i][1])
+ }
+
+ def distinctCount = sql "SELECT count(distinct id) FROM ${tableName}"
+ Assert.assertEquals(5L, (distinctCount[0][0] as Number).longValue())
+
+ def sumId = sql "SELECT sum(id) FROM ${tableName}"
+ Assert.assertEquals(15L, (sumId[0][0] as Number).longValue())
+
+ def finalState = sql "SHOW ROUTINE LOAD FOR ${job}"
+ Assert.assertEquals("RUNNING", finalState[0][8].toString())
+ } finally {
+ try {
+ sql "STOP ROUTINE LOAD FOR ${job}"
+ } catch (Exception e) {
+ logger.warn("Failed to stop routine load: {}", e.message)
+ }
+ producer.close()
+ adminClient.close()
+ sql "DROP TABLE IF EXISTS ${tableName}"
+ }
+ }
+}
diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load_same_job_schema_replay_exactly_once.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load_same_job_schema_replay_exactly_once.groovy
new file mode 100644
index 00000000000..e6b148dcbf6
--- /dev/null
+++ b/regression-test/suites/load_p0/routine_load/test_routine_load_same_job_schema_replay_exactly_once.groovy
@@ -0,0 +1,215 @@
+// 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.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.RoutineLoadTestUtils
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.junit.Assert
+
+// 语义说明：
+// Routine Load 因 strict_mode/schema mismatch 暂停后，RESUME 不会回放历史坏数据，
+// 只会继续消费恢复后追加的新数据。
+suite("test_routine_load_resume_without_replay_after_schema_pause", "docker") {
+ def topicSuffix = System.currentTimeMillis()
+ def kafkaCsvTopic = "test_rl_resume_no_replay_${topicSuffix}"
+
+ if (!RoutineLoadTestUtils.isKafkaTestEnabled(context)) {
+ return
+ }
+
+ def kafkaBroker = RoutineLoadTestUtils.getKafkaBroker(context)
+
+ def options = new ClusterOptions()
+ options.setFeNum(1)
+ options.setBeNum(1)
+ options.cloudMode = false
+
+ docker(options) {
+ def producer = RoutineLoadTestUtils.createKafkaProducer(kafkaBroker)
+
+ def tableName = "test_rl_resume_no_replay"
+ def job = "test_rl_resume_no_replay_job_${topicSuffix}"
+
+ def schemaRejectedData = [
+ "1,aaa,2023-01-01,val1,2023-01-01 10:00:00,abcdef",
+ "2,bbb,2023-01-02,val2,2023-01-02 11:00:00,ghijkl"
+ ]
+ def postResumeData = [
+ "3,ccc,2023-01-03,val3,2023-01-03 12:00:00,hello1",
+ "4,ddd,2023-01-04,val4,2023-01-04 13:00:00,world2"
+ ]
+
+ def waitForPaused = { int maxWait = 60 ->
+ def waited = 0
+ while (waited < maxWait) {
+ def res = sql "SHOW ROUTINE LOAD FOR ${job}"
+ def state = res[0][8].toString()
+ if (state == "PAUSED") {
+ logger.info("Job paused as expected. reason={}", res[0][17].toString())
+ return
+ }
+ sleep(1000)
+ waited++
+ }
+ Assert.fail("Routine load job did not enter PAUSED within timeout")
+ }
+
+ def waitForSchemaChangeFinished = { int maxWait = 120 ->
+ def waited = 0
+ while (waited < maxWait) {
+ def alterRes = sql "SHOW ALTER TABLE COLUMN WHERE TableName='${tableName}' ORDER BY CreateTime DESC LIMIT 1"
+ if (alterRes.size() > 0 && alterRes[0][9].toString() == "FINISHED") {
+ return
+ }
+ sleep(1000)
+ waited++
+ }
+ Assert.fail("Schema change did not finish within timeout")
+ }
+
+ def waitForExactRowCount = { long expected, int maxWait = 120 ->
+ def waited = 0
+ while (waited < maxWait) {
+ def jobRes = sql "SHOW ROUTINE LOAD FOR ${job}"
+ def state = jobRes[0][8].toString()
+ def countRes = sql "SELECT count(*) FROM ${tableName}"
+ long rowCount = (countRes[0][0] as Number).longValue()
+ logger.info("waitForExactRowCount: state={}, rows={}", state, rowCount)
+ if (state == "RUNNING" && rowCount == expected) {
+ return
+ }
+ sleep(1000)
+ waited++
+ }
+ Assert.fail("Row count did not reach ${expected} within timeout")
+ }
+
+ def waitForRunning = { int maxWait = 60 ->
+ def waited = 0
+ while (waited < maxWait) {
+ def jobRes = sql "SHOW ROUTINE LOAD FOR ${job}"
+ def state = jobRes[0][8].toString()
+ if (state == "RUNNING") {
+ return
+ }
+ sleep(1000)
+ waited++
+ }
+ Assert.fail("Routine load job did not enter RUNNING within timeout")
+ }
+
+ def assertRowCountKeeps = { long expected, int observeSeconds = 5 ->
+ for (int i = 0; i < observeSeconds; i++) {
+ def countRes = sql "SELECT count(*) FROM ${tableName}"
+ long rowCount = (countRes[0][0] as Number).longValue()
+ logger.info("assertRowCountKeeps: second={}, rows={}", i, rowCount)
+ Assert.assertEquals("Observed row count should stay at ${expected}", expected, rowCount)
+ sleep(1000)
+ }
+ }
+
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName} (
+ `k1` int(20) NULL,
+ `k2` string NULL,
+ `v1` date NULL,
+ `v2` string NULL,
+ `v3` datetime NULL,
+ `v4` varchar(5) NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`k1`)
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 1
+ PROPERTIES ("replication_allocation" = "tag.location.default: 1");
+ """
+ sql "sync"
+
+ schemaRejectedData.each { line ->
+ logger.info("Sending schema-blocked row to Kafka: {}", line)
+ producer.send(new ProducerRecord<>(kafkaCsvTopic, null, line)).get()
+ }
+ producer.flush()
+
+ try {
+ sql """
+ CREATE ROUTINE LOAD ${job} ON ${tableName}
+ COLUMNS TERMINATED BY ","
+ PROPERTIES
+ (
+ "strict_mode" = "true",
+ "max_batch_interval" = "5",
+ "max_batch_rows" = "300000",
+ "max_batch_size" = "209715200"
+ )
+ FROM KAFKA
+ (
+ "kafka_broker_list" = "${kafkaBroker}",
+ "kafka_topic" = "${kafkaCsvTopic}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING"
+ );
+ """
+ sql "sync"
+
+ waitForPaused()
+
+ def pausedCount = sql "SELECT count(*) FROM ${tableName}"
+ Assert.assertEquals("PAUSED 期间不应写入任何行", 0L, (pausedCount[0][0] as Number).longValue())
+
+ sql "ALTER TABLE ${tableName} MODIFY COLUMN v4 VARCHAR(10)"
+ waitForSchemaChangeFinished()
+
+ sql "RESUME ROUTINE LOAD FOR ${job}"
+ waitForRunning()
+ assertRowCountKeeps(0L)
+
+ // schema 放宽并 RESUME 后，同一个 job 不会补消费旧坏数据，
+ // 但应继续消费恢复后追加的新数据。
+ postResumeData.each { line ->
+ logger.info("Sending post-resume row to Kafka: {}", line)
+ producer.send(new ProducerRecord<>(kafkaCsvTopic, null, line)).get()
+ }
+ producer.flush()
+ waitForExactRowCount(2L)
+
+ def finalRows = sql "SELECT k1, k2, v4 FROM ${tableName} ORDER BY k1"
+ Assert.assertEquals("resume 后应只消费 schema change 之后追加的 2 行", 2, finalRows.size())
+ Assert.assertEquals(3, finalRows[0][0])
+ Assert.assertEquals("ccc", finalRows[0][1])
+ Assert.assertEquals("hello1", finalRows[0][2])
+ Assert.assertEquals(4, finalRows[1][0])
+ Assert.assertEquals("ddd", finalRows[1][1])
+ Assert.assertEquals("world2", finalRows[1][2])
+
+ def oldRows = sql "SELECT count(*) FROM ${tableName} WHERE k1 IN (1, 2)"
+ Assert.assertEquals("resume 后不应回放历史 schema-failed 数据", 0L, (oldRows[0][0] as Number).longValue())
+
+ def sumKey = sql "SELECT sum(k1) FROM ${tableName}"
+ Assert.assertEquals(7L, (sumKey[0][0] as Number).longValue())
+
+ def finalState = sql "SHOW ROUTINE LOAD FOR ${job}"
+ Assert.assertEquals("RUNNING", finalState[0][8].toString())
+ } finally {
+ try {
+ sql "STOP ROUTINE LOAD FOR ${job}"
+ } catch (Exception e) {
+ logger.warn("Failed to stop routine load: {}", e.message)
+ }
+ producer.close()
+ sql "DROP TABLE IF EXISTS ${tableName}"
+ }
+ }
+}
diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load_too_many_tasks_recovery_exactly_once.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load_too_many_tasks_recovery_exactly_once.groovy
new file mode 100644
index 00000000000..312f0e794eb
--- /dev/null
+++ b/regression-test/suites/load_p0/routine_load/test_routine_load_too_many_tasks_recovery_exactly_once.groovy
@@ -0,0 +1,177 @@
+// 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.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.RoutineLoadTestUtils
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.junit.Assert
+
+suite("test_routine_load_too_many_tasks_recovery_exactly_once", "docker") {
+ def topicSuffix = System.currentTimeMillis()
+ def kafkaCsvTopic = "test_rl_too_many_tasks_recovery_${topicSuffix}"
+
+ if (!RoutineLoadTestUtils.isKafkaTestEnabled(context)) {
+ return
+ }
+
+ def kafkaPort = context.config.otherConfigs.get("kafka_port")
+ Assert.assertNotNull("kafka_port must be configured in regression-conf.groovy", kafkaPort)
+ def kafka_broker = "127.0.0.1:${kafkaPort}"
+
+ def options = new ClusterOptions()
+ options.setFeNum(1)
+ options.setBeNum(1)
+ options.cloudMode = false
+ options.enableDebugPoints()
+
+ docker(options) {
+ def producer = RoutineLoadTestUtils.createKafkaProducer(kafka_broker)
+
+ def tableName = "test_rl_too_many_tasks_recovery"
+ def job = "test_too_many_tasks_recovery_job_${topicSuffix}"
+ def debugPoint = "FE.ROUTINE_LOAD_TASK_SUBMIT_FAILED.TOO_MANY_TASKS"
+
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName} (
+ `id` int NOT NULL,
+ `payload` string NULL,
+ `ts` string NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`id`)
+ DISTRIBUTED BY HASH(`id`) BUCKETS 1
+ PROPERTIES ("replication_allocation" = "tag.location.default: 1");
+ """
+ sql "sync"
+
+ List<String> testData = []
+ for (int i = 1; i <= 20; i++) {
+ testData.add("${i},payload_${i},2026-02-${String.format('%02d', i)}".toString())
+ }
+
+ testData.each { line ->
+ logger.info("Sending to Kafka: ${line}")
+ def record = new ProducerRecord<>(kafkaCsvTopic, null, line.toString())
+ producer.send(record).get()
+ }
+ producer.flush()
+ logger.info("All 20 records sent to Kafka")
+
+ try {
+ GetDebugPoint().enableDebugPointForAllFEs(debugPoint)
+ logger.info("Debug point enabled: ${debugPoint}")
+
+ sql """
+ CREATE ROUTINE LOAD ${job} ON ${tableName}
+ COLUMNS TERMINATED BY ","
+ PROPERTIES
+ (
+ "max_batch_interval" = "5"
+ )
+ FROM KAFKA
+ (
+ "kafka_broker_list" = "${kafka_broker}",
+ "kafka_topic" = "${kafkaCsvTopic}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING"
+ );
+ """
+
+ def waitCount = 0
+ def sawFailure = false
+ while (waitCount < 30) {
+ sleep(1000)
+ def state = sql "show routine load for ${job}"
+ def routineLoadState = state[0][8].toString()
+ def statistic = state[0][14].toString()
+ def otherMsg = state[0][19].toString()
+ logger.info("State: ${routineLoadState}, stats: ${statistic}, msg: ${otherMsg}")
+
+ def statJson = new groovy.json.JsonSlurper().parseText(statistic)
+ long abortedCount = statJson.abortedTaskNum as long
+ if (abortedCount > 0 || otherMsg.contains("TOO_MANY_TASKS")) {
+ sawFailure = true
+ logger.info("Detected TOO_MANY_TASKS failure: abortedTaskNum=${abortedCount}, msg=${otherMsg}")
+ break
+ }
+ waitCount++
+ }
+ Assert.assertTrue("应观察到 TOO_MANY_TASKS 失败（abortedTaskNum>0 或 otherMsg 含相关信息）", sawFailure)
+
+ def failPhaseCount = sql "select count(*) from ${tableName}"
+ logger.info("Row count during failure phase: ${failPhaseCount[0][0]}")
+
+ GetDebugPoint().disableDebugPointForAllFEs(debugPoint)
+ logger.info("Debug point disabled, waiting for recovery...")
+
+ def maxWaitCount = 120
+ def count = 0
+ while (true) {
+ def state = sql "show routine load for ${job}"
+ def routineLoadState = state[0][8].toString()
+ def statistic = state[0][14].toString()
+ logger.info("Recovery phase - State: ${routineLoadState}, stats: ${statistic}")
+
+ def rowCount = sql "select count(*) from ${tableName}"
+ if (routineLoadState == "RUNNING" && rowCount[0][0] == 20) {
+ logger.info("Job recovered and all 20 rows consumed")
+ break
+ }
+ if (count > maxWaitCount) {
+ Assert.fail("Routine load did not recover within timeout. "
+ + "State: ${routineLoadState}, rows: ${rowCount[0][0]}")
+ }
+ sleep(1000)
+ count++
+ }
+
+ def finalCount = sql "select count(*) from ${tableName}"
+ Assert.assertEquals("应精确消费 20 行", 20L, finalCount[0][0])
+
+ def distinctCount = sql "select count(distinct id) from ${tableName}"
+ Assert.assertEquals("不应有重复 id", 20L, distinctCount[0][0])
+
+ def sumId = sql "select sum(id) from ${tableName}"
+ Assert.assertEquals("sum(id) 应为 1+2+...+20=210", 210L, sumId[0][0])
+
+ def minId = sql "select min(id) from ${tableName}"
+ Assert.assertEquals("min(id) 应为 1", 1, minId[0][0])
+
+ def maxId = sql "select max(id) from ${tableName}"
+ Assert.assertEquals("max(id) 应为 20", 20, maxId[0][0])
+
+ def finalState = sql "show routine load for ${job}"
+ Assert.assertEquals("最终状态应为 RUNNING", "RUNNING", finalState[0][8].toString())
+ logger.info("All exactly-once assertions passed after TOO_MANY_TASKS recovery")
+ } catch (Exception e) {
+ logger.error("Test failed: ${e.getMessage()}")
+ throw e
+ } finally {
+ try {
+ GetDebugPoint().disableDebugPointForAllFEs(debugPoint)
+ } catch (Exception e) {
+ logger.warn("Failed to disable debug point in cleanup: ${e.message}")
+ }
+ try {
+ sql "stop routine load for ${job}"
+ } catch (Exception e) {
+ logger.warn("Failed to stop routine load: ${e.message}")
+ }
+ producer.close()
+ sql "DROP TABLE IF EXISTS ${tableName}"
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org