From f951d0aa39b87fd190251d4dae8c4d25433a4d4f Mon Sep 17 00:00:00 2001 From: laihui Date: Mon, 13 Jul 2026 23:13:21 +0800 Subject: [PATCH 01/11] show first error message for routine load --- .../schema_routine_load_job_scanner.cpp | 8 +- .../load/stream_load/stream_load_executor.cpp | 5 +- .../org/apache/doris/catalog/SchemaTable.java | 1 + .../doris/cloud/transaction/TxnUtil.java | 3 + .../RLTaskTxnCommitAttachment.java | 9 ++ .../load/routineload/RoutineLoadJob.java | 13 ++ .../commands/ShowRoutineLoadCommand.java | 1 + .../doris/service/FrontendServiceImpl.java | 1 + .../load/routineload/RoutineLoadJobTest.java | 30 ++++ .../commands/ShowRoutineLoadCommandTest.java | 2 + gensrc/proto/cloud.proto | 1 + gensrc/thrift/FrontendService.thrift | 2 + .../test_routine_load_first_error_msg.groovy | 129 ++++++++++++++++++ 13 files changed, 202 insertions(+), 3 deletions(-) create mode 100644 regression-test/suites/load_p0/routine_load/test_routine_load_first_error_msg.groovy diff --git a/be/src/information_schema/schema_routine_load_job_scanner.cpp b/be/src/information_schema/schema_routine_load_job_scanner.cpp index 9e94a507d6c98a..345d278619d75c 100644 --- a/be/src/information_schema/schema_routine_load_job_scanner.cpp +++ b/be/src/information_schema/schema_routine_load_job_scanner.cpp @@ -51,6 +51,7 @@ std::vector SchemaRoutineLoadJobScanner::_s_tbls_colu {"LAG", TYPE_STRING, sizeof(StringRef), true}, {"REASON_OF_STATE_CHANGED", TYPE_STRING, sizeof(StringRef), true}, {"ERROR_LOG_URLS", TYPE_STRING, sizeof(StringRef), true}, + {"FIRST_ERROR_MSG", TYPE_STRING, sizeof(StringRef), true}, {"USER_NAME", TYPE_STRING, sizeof(StringRef), true}, {"CURRENT_ABORT_TASK_NUM", TYPE_INT, sizeof(int32_t), true}, {"IS_ABNORMAL_PAUSE", TYPE_BOOLEAN, sizeof(int8_t), true}, @@ -169,10 +170,13 @@ Status SchemaRoutineLoadJobScanner::_fill_block_impl(Block* block) { case 16: // ERROR_LOG_URLS column_value = job_info.__isset.error_log_urls ? job_info.error_log_urls : ""; break; - case 17: // USER_NAME + case 17: // FIRST_ERROR_MSG + column_value = job_info.__isset.first_error_msg ? job_info.first_error_msg : ""; + break; + case 18: // USER_NAME column_value = job_info.__isset.user_name ? job_info.user_name : ""; break; - case 20: // COMPUTE_GROUP + case 21: // COMPUTE_GROUP column_value = job_info.__isset.compute_group ? job_info.compute_group : ""; break; } diff --git a/be/src/load/stream_load/stream_load_executor.cpp b/be/src/load/stream_load/stream_load_executor.cpp index 478ed22755df9e..66c1d7fd339f30 100644 --- a/be/src/load/stream_load/stream_load_executor.cpp +++ b/be/src/load/stream_load/stream_load_executor.cpp @@ -99,6 +99,7 @@ Status StreamLoadExecutor::execute_plan_fragment( ctx->loaded_bytes = state->num_bytes_load_total(); int64_t num_selected_rows = ctx->number_total_rows - ctx->number_unselected_rows; ctx->error_url = to_load_error_http_path(state->get_error_log_file_path()); + ctx->first_error_msg = state->get_first_error_msg(); if (status->ok() && !ctx->group_commit && num_selected_rows > 0 && (double)ctx->number_filtered_rows / num_selected_rows > ctx->max_filter_ratio) { // NOTE: Do not modify the error message here, for historical reasons, @@ -118,7 +119,6 @@ Status StreamLoadExecutor::execute_plan_fragment( LOG(WARNING) << "fragment execute failed" << ", err_msg=" << status->to_string() << ", " << ctx->brief(); ctx->number_loaded_rows = 0; - ctx->first_error_msg = state->get_first_error_msg(); // cancel body_sink, make sender known it if (ctx->body_sink != nullptr) { ctx->body_sink->cancel(status->to_string()); @@ -428,6 +428,9 @@ bool StreamLoadExecutor::collect_load_stat(StreamLoadContext* ctx, TTxnCommitAtt rl_attach.__set_receivedBytes(ctx->receive_bytes); rl_attach.__set_loadedBytes(ctx->loaded_bytes); rl_attach.__set_loadCostMs(ctx->load_cost_millis); + if (!ctx->first_error_msg.empty()) { + rl_attach.__set_firstErrorMsg(ctx->first_error_msg); + } attach->rlTaskTxnCommitAttachment = rl_attach; attach->__isset.rlTaskTxnCommitAttachment = true; diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java index fdcb838ef6f685..d70fa9ed9f98ed 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java @@ -702,6 +702,7 @@ public class SchemaTable extends Table { .column("LAG", ScalarType.createStringType()) .column("REASON_OF_STATE_CHANGED", ScalarType.createStringType()) .column("ERROR_LOG_URLS", ScalarType.createStringType()) + .column("FIRST_ERROR_MSG", ScalarType.createStringType()) .column("USER_NAME", ScalarType.createStringType()) .column("CURRENT_ABORT_TASK_NUM", ScalarType.createType(PrimitiveType.INT)) .column("IS_ABNORMAL_PAUSE", ScalarType.createType(PrimitiveType.BOOLEAN)) diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/TxnUtil.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/TxnUtil.java index f29aa5294b3781..fceb16d91add52 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/TxnUtil.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/TxnUtil.java @@ -256,6 +256,9 @@ public static TxnCommitAttachmentPB rlTaskTxnCommitAttachmentToPb(RLTaskTxnCommi if (rtTaskTxnCommitAttachment.getErrorLogUrl() != null) { builder.setErrorLogUrl(rtTaskTxnCommitAttachment.getErrorLogUrl()); } + if (rtTaskTxnCommitAttachment.getFirstErrorMsg() != null) { + builder.setFirstErrorMsg(rtTaskTxnCommitAttachment.getFirstErrorMsg()); + } attachementBuilder.setRlTaskTxnCommitAttachment(builder.build()); return attachementBuilder.build(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RLTaskTxnCommitAttachment.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RLTaskTxnCommitAttachment.java index dd9c4ed85ce908..71d85c264ec447 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RLTaskTxnCommitAttachment.java +++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RLTaskTxnCommitAttachment.java @@ -46,6 +46,7 @@ public class RLTaskTxnCommitAttachment extends TxnCommitAttachment { @SerializedName(value = "pro") private RoutineLoadProgress progress; private String errorLogUrl; + private String firstErrorMsg; public RLTaskTxnCommitAttachment() { super(TransactionState.LoadJobSourceType.ROUTINE_LOAD_TASK); @@ -75,6 +76,9 @@ public RLTaskTxnCommitAttachment(TRLTaskTxnCommitAttachment rlTaskTxnCommitAttac if (rlTaskTxnCommitAttachment.isSetErrorLogUrl()) { this.errorLogUrl = rlTaskTxnCommitAttachment.getErrorLogUrl(); } + if (rlTaskTxnCommitAttachment.isSetFirstErrorMsg()) { + this.firstErrorMsg = rlTaskTxnCommitAttachment.getFirstErrorMsg(); + } } public RLTaskTxnCommitAttachment(RLTaskTxnCommitAttachmentPB rlTaskTxnCommitAttachment) { @@ -92,6 +96,7 @@ public RLTaskTxnCommitAttachment(RLTaskTxnCommitAttachmentPB rlTaskTxnCommitAtta this.progress = progress; this.errorLogUrl = rlTaskTxnCommitAttachment.getErrorLogUrl(); + this.firstErrorMsg = rlTaskTxnCommitAttachment.getFirstErrorMsg(); } public long getJobId() { @@ -134,6 +139,10 @@ public String getErrorLogUrl() { return errorLogUrl; } + public String getFirstErrorMsg() { + return firstErrorMsg; + } + @Override public String toString() { return "RLTaskTxnCommitAttachment [filteredRows=" + filteredRows 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 3223ff913594e1..8bc795f32e472f 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 @@ -276,6 +276,8 @@ public boolean isFinalState() { // save the latest 3 error log urls private Queue errorLogUrls = EvictingQueue.create(3); + @SerializedName("fem") + private String firstErrorMsg = ""; @SerializedName("ccid") private String cloudClusterId; @@ -809,6 +811,10 @@ public Queue getErrorLogUrls() { return errorLogUrls; } + public String getFirstErrorMsg() { + return Strings.nullToEmpty(firstErrorMsg); + } + // RoutineLoadScheduler will run this method at fixed interval, and renew the timeout tasks public void processTimeoutTasks() { writeLock(); @@ -1408,6 +1414,12 @@ private void executeTaskOnTxnStatusChanged(RoutineLoadTaskInfo routineLoadTaskIn routineLoadTaskInfo.handleTaskByTxnCommitAttachment(rlTaskTxnCommitAttachment); } + if (rlTaskTxnCommitAttachment != null + && !Strings.isNullOrEmpty(rlTaskTxnCommitAttachment.getFirstErrorMsg())) { + firstErrorMsg = StringUtils.abbreviate( + rlTaskTxnCommitAttachment.getFirstErrorMsg(), Config.first_error_msg_max_length); + } + if (rlTaskTxnCommitAttachment != null && !Strings.isNullOrEmpty(rlTaskTxnCommitAttachment.getErrorLogUrl())) { errorLogUrls.add(rlTaskTxnCommitAttachment.getErrorLogUrl()); } @@ -1712,6 +1724,7 @@ public List getShowInfo() { row.add(userIdentity.getQualifiedUser()); row.add(comment); row.add(getClusterInfo()); + row.add(getFirstErrorMsg()); return row; } finally { readUnlock(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowRoutineLoadCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowRoutineLoadCommand.java index fe5c1441ebc6d6..42737314470671 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowRoutineLoadCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowRoutineLoadCommand.java @@ -110,6 +110,7 @@ public class ShowRoutineLoadCommand extends ShowCommand { .add("User") .add("Comment") .add("ComputeGroup") + .add("FirstErrorMsg") .build(); private final LabelNameInfo labelNameInfo; diff --git a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java index f76b687e77fb00..afd0ed46b2a3dd 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java +++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java @@ -5582,6 +5582,7 @@ public TFetchRoutineLoadJobResult fetchRoutineLoadJob(TFetchRoutineLoadJobReques jobInfo.setLag(job.getLag()); jobInfo.setReasonOfStateChanged(job.getStateReason()); jobInfo.setErrorLogUrls(Joiner.on(", ").join(job.getErrorLogUrls())); + jobInfo.setFirstErrorMsg(job.getFirstErrorMsg()); jobInfo.setUserName(job.getUserIdentity().getQualifiedUser()); jobInfo.setCurrentAbortTaskNum(job.getJobStatistic().currentAbortedTaskNum); jobInfo.setIsAbnormalPause(job.isAbnormalPause()); diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java index 8876c6a8aea9bc..4bedab14b7fe1e 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java @@ -21,6 +21,7 @@ import org.apache.doris.catalog.Database; import org.apache.doris.catalog.Env; import org.apache.doris.catalog.Table; +import org.apache.doris.cloud.transaction.TxnUtil; import org.apache.doris.common.InternalErrorCode; import org.apache.doris.common.Pair; import org.apache.doris.common.UserException; @@ -33,6 +34,9 @@ import org.apache.doris.nereids.trees.plans.commands.info.CreateRoutineLoadInfo; import org.apache.doris.persist.EditLog; import org.apache.doris.thrift.TKafkaRLTaskProgress; +import org.apache.doris.thrift.TLoadSourceType; +import org.apache.doris.thrift.TRLTaskTxnCommitAttachment; +import org.apache.doris.thrift.TUniqueId; import org.apache.doris.thrift.TUniqueKeyUpdateMode; import org.apache.doris.transaction.GlobalTransactionMgrIface; import org.apache.doris.transaction.TransactionException; @@ -53,6 +57,25 @@ import java.util.Map; public class RoutineLoadJobTest { + @Test + public void testFirstErrorMsgInTxnCommitAttachment() { + TRLTaskTxnCommitAttachment thriftAttachment = new TRLTaskTxnCommitAttachment(); + thriftAttachment.setLoadSourceType(TLoadSourceType.KAFKA); + thriftAttachment.setId(new TUniqueId(1, 2)); + thriftAttachment.setJobId(3); + thriftAttachment.setKafkaRLTaskProgress(new TKafkaRLTaskProgress(Maps.newHashMap())); + thriftAttachment.setErrorLogUrl("http://127.0.0.1/error_log"); + thriftAttachment.setFirstErrorMsg("invalid source row"); + + RLTaskTxnCommitAttachment attachment = new RLTaskTxnCommitAttachment(thriftAttachment); + Assert.assertEquals("invalid source row", attachment.getFirstErrorMsg()); + + RLTaskTxnCommitAttachment cloudAttachment = TxnUtil.rtTaskTxnCommitAttachmentFromPb( + TxnUtil.rlTaskTxnCommitAttachmentToPb(attachment)); + Assert.assertEquals("http://127.0.0.1/error_log", cloudAttachment.getErrorLogUrl()); + Assert.assertEquals("invalid source row", cloudAttachment.getFirstErrorMsg()); + } + @Test public void testAfterAbortedReasonOffsetOutOfRange() throws UserException { Env env = Mockito.mock(Env.class); @@ -95,6 +118,8 @@ public void testAfterAborted() throws UserException { tKafkaRLTaskProgress.partitionCmtOffset = Maps.newHashMap(); KafkaProgress kafkaProgress = new KafkaProgress(tKafkaRLTaskProgress); Deencapsulation.setField(attachment, "progress", kafkaProgress); + Deencapsulation.setField(attachment, "errorLogUrl", "http://127.0.0.1/error_log"); + Deencapsulation.setField(attachment, "firstErrorMsg", "invalid source row"); KafkaProgress currentProgress = new KafkaProgress(tKafkaRLTaskProgress); @@ -114,6 +139,8 @@ public void testAfterAborted() throws UserException { Assert.assertEquals(RoutineLoadJob.JobState.RUNNING, routineLoadJob.getState()); Assert.assertEquals(new Long(1), Deencapsulation.getField(jobStatistic, "abortedTaskNum")); + Assert.assertEquals("http://127.0.0.1/error_log", routineLoadJob.getErrorLogUrls().peek()); + Assert.assertEquals("invalid source row", routineLoadJob.getFirstErrorMsg()); } @Test @@ -153,10 +180,13 @@ public void testGetShowInfo() { Deencapsulation.setField(routineLoadJob, "pauseReason", errorReason); Deencapsulation.setField(routineLoadJob, "progress", kafkaProgress); Deencapsulation.setField(routineLoadJob, "userIdentity", userIdentity); + Deencapsulation.setField(routineLoadJob, "firstErrorMsg", "invalid source row"); List showInfo = routineLoadJob.getShowInfo(); Assert.assertEquals(true, showInfo.stream().filter(entity -> !Strings.isNullOrEmpty(entity)) .anyMatch(entity -> entity.equals(errorReason.toString()))); + Assert.assertEquals(24, showInfo.size()); + Assert.assertEquals("invalid source row", showInfo.get(23)); } @Test diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ShowRoutineLoadCommandTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ShowRoutineLoadCommandTest.java index 4bbe272203040a..ce933be3decb99 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ShowRoutineLoadCommandTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ShowRoutineLoadCommandTest.java @@ -35,5 +35,7 @@ public void testValidate() { LabelNameInfo labelNameInfo = new LabelNameInfo("test_db", "test_label"); ShowRoutineLoadCommand command = new ShowRoutineLoadCommand(labelNameInfo, null, false); Assertions.assertDoesNotThrow(() -> command.validate(connectContext)); + Assertions.assertEquals(24, command.getMetaData().getColumnCount()); + Assertions.assertEquals("FirstErrorMsg", command.getMetaData().getColumn(23).getName()); } } diff --git a/gensrc/proto/cloud.proto b/gensrc/proto/cloud.proto index bff85eee942756..6c85e190c01e4e 100644 --- a/gensrc/proto/cloud.proto +++ b/gensrc/proto/cloud.proto @@ -406,6 +406,7 @@ message RLTaskTxnCommitAttachmentPB { optional int64 task_execution_time_ms = 7; optional RoutineLoadProgressPB progress = 8; optional string error_log_url = 9; + optional string first_error_msg = 10; } message RoutineLoadJobStatisticPB { diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift index d87d7df56edd92..2bc093da9b9f20 100644 --- a/gensrc/thrift/FrontendService.thrift +++ b/gensrc/thrift/FrontendService.thrift @@ -675,6 +675,7 @@ struct TRLTaskTxnCommitAttachment { 10: optional TKafkaRLTaskProgress kafkaRLTaskProgress 11: optional string errorLogUrl 12: optional TKinesisRLTaskProgress kinesisRLTaskProgress + 13: optional string firstErrorMsg } struct TTxnCommitAttachment { @@ -1668,6 +1669,7 @@ struct TRoutineLoadJob { 19: optional i32 current_abort_task_num 20: optional bool is_abnormal_pause 21: optional string compute_group + 22: optional string first_error_msg } struct TFetchRoutineLoadJobResult { diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load_first_error_msg.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load_first_error_msg.groovy new file mode 100644 index 00000000000000..5fea612394daa8 --- /dev/null +++ b/regression-test/suites/load_p0/routine_load/test_routine_load_first_error_msg.groovy @@ -0,0 +1,129 @@ +// 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.admin.AdminClient +import org.apache.kafka.clients.admin.NewTopic +import org.apache.kafka.clients.producer.ProducerConfig +import org.junit.Assert + +suite("test_routine_load_first_error_msg", "p0") { + if (!RoutineLoadTestUtils.isKafkaTestEnabled(context)) { + return + } + + def kafkaBroker = RoutineLoadTestUtils.getKafkaBroker(context) + def kafkaTopic = "test_routine_load_first_error_msg_${System.currentTimeMillis()}" + def jobName = "test_routine_load_first_error_msg" + def tableName = "test_routine_load_first_error_msg" + + def adminProps = new Properties() + adminProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBroker) + def adminClient = AdminClient.create(adminProps) + try { + adminClient.createTopics([new NewTopic(kafkaTopic, 1, (short) 1)]).all().get() + } finally { + adminClient.close() + } + + try { + sql "DROP TABLE IF EXISTS ${tableName}" + sql """ + CREATE TABLE ${tableName} ( + id INT, + name STRING + ) + PARTITION BY RANGE(id) ( + PARTITION p0 VALUES LESS THAN ("10") + ) + DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ("replication_num" = "1") + """ + + sql """ + CREATE ROUTINE LOAD ${jobName} ON ${tableName} + COLUMNS TERMINATED BY "|" + PROPERTIES ( + "max_error_number" = "10", + "max_filter_ratio" = "1.0", + "max_batch_interval" = "5" + ) + FROM KAFKA ( + "kafka_broker_list" = "${kafkaBroker}", + "kafka_topic" = "${kafkaTopic}", + "property.kafka_default_offsets" = "OFFSET_END" + ) + """ + + def count = 0 + while (true) { + def showResult = sql "SHOW ROUTINE LOAD FOR ${jobName}" + if (showResult[0][8].toString() == "RUNNING") { + break + } + if (count++ > 60) { + Assert.fail("Routine Load job did not enter RUNNING state") + } + sleep(1000) + } + + def producer = RoutineLoadTestUtils.createKafkaProducer(kafkaBroker) + try { + RoutineLoadTestUtils.sendTestDataToKafka( + producer, [kafkaTopic], ["100|bad_row", "1|valid_row"]) + producer.flush() + } finally { + producer.close() + } + + count = 0 + while (true) { + def jobInfo = sql """ + SELECT ERROR_LOG_URLS, FIRST_ERROR_MSG + FROM information_schema.routine_load_jobs + WHERE JOB_NAME = '${jobName}' + """ + def showResult = sql "SHOW ROUTINE LOAD FOR ${jobName}" + def loadedRows = sql "SELECT COUNT(*) FROM ${tableName}" + def informationSchemaErrorLogUrls = jobInfo[0][0].toString() + def firstErrorMsg = jobInfo[0][1] + def showErrorLogUrls = showResult[0][18].toString() + def showFirstErrorMsg = showResult[0][23] + if (loadedRows[0][0] == 1 && firstErrorMsg != null + && firstErrorMsg.toString().contains("100|bad_row") + && showFirstErrorMsg != null + && showFirstErrorMsg.toString().contains("100|bad_row")) { + assertEquals(informationSchemaErrorLogUrls, showErrorLogUrls) + assertFalse(informationSchemaErrorLogUrls.contains("first_error_msg:")) + assertFalse(showErrorLogUrls.contains("first_error_msg:")) + assertEquals(firstErrorMsg.toString(), showFirstErrorMsg.toString()) + break + } + if (count++ > 60) { + Assert.fail("First error message was not reported by routine_load_jobs and SHOW ROUTINE LOAD: " + + "loadedRows=${loadedRows[0][0]}, firstErrorMsg=${firstErrorMsg}, " + + "showFirstErrorMsg=${showFirstErrorMsg}, " + + "informationSchemaErrorLogUrlsEmpty=${informationSchemaErrorLogUrls.isEmpty()}, " + + "showErrorLogUrlsEmpty=${showErrorLogUrls.isEmpty()}") + } + sleep(1000) + } + } finally { + try_sql "STOP ROUTINE LOAD FOR ${jobName}" + try_sql "DROP TABLE IF EXISTS ${tableName}" + } +} From 103b1466aae01738117ed5c00f071236cb80b2d3 Mon Sep 17 00:00:00 2001 From: laihui Date: Tue, 14 Jul 2026 00:19:55 +0800 Subject: [PATCH 02/11] [fix](load) Preserve routine load first error messages ### What problem does this PR solve? Related PR: #65553 Problem Summary: Routine Load first error display fields were not restored during transaction replay, and cloud progress persistence dropped the first error message during meta-service refresh. The information_schema column also shifted existing ordinals, and the regression test did not follow local table naming and cleanup conventions. This change restores display fields independently from progress validation, persists the latest non-empty first error message in cloud progress, appends the information_schema field, and updates replay, cloud, and regression coverage. ### Release note None ### Check List (For Author) - Test: Not run per request; git diff --check passed - Behavior changed: Yes, Routine Load retains the first error message across FE replay and cloud progress refresh - Does this need documentation: No --- .../schema_routine_load_job_scanner.cpp | 14 ++-- cloud/src/meta-service/meta_service_txn.cpp | 8 +++ cloud/test/meta_service_job_test.cpp | 65 +++++++++++++++++++ .../org/apache/doris/catalog/SchemaTable.java | 2 +- .../load/routineload/RoutineLoadJob.java | 41 ++++++++---- .../load/routineload/RoutineLoadJobTest.java | 42 ++++++++++++ gensrc/proto/cloud.proto | 1 + .../test_routine_load_first_error_msg.groovy | 10 ++- 8 files changed, 155 insertions(+), 28 deletions(-) diff --git a/be/src/information_schema/schema_routine_load_job_scanner.cpp b/be/src/information_schema/schema_routine_load_job_scanner.cpp index 345d278619d75c..d2d54b60911e28 100644 --- a/be/src/information_schema/schema_routine_load_job_scanner.cpp +++ b/be/src/information_schema/schema_routine_load_job_scanner.cpp @@ -51,11 +51,11 @@ std::vector SchemaRoutineLoadJobScanner::_s_tbls_colu {"LAG", TYPE_STRING, sizeof(StringRef), true}, {"REASON_OF_STATE_CHANGED", TYPE_STRING, sizeof(StringRef), true}, {"ERROR_LOG_URLS", TYPE_STRING, sizeof(StringRef), true}, - {"FIRST_ERROR_MSG", TYPE_STRING, sizeof(StringRef), true}, {"USER_NAME", TYPE_STRING, sizeof(StringRef), true}, {"CURRENT_ABORT_TASK_NUM", TYPE_INT, sizeof(int32_t), true}, {"IS_ABNORMAL_PAUSE", TYPE_BOOLEAN, sizeof(int8_t), true}, {"COMPUTE_GROUP", TYPE_STRING, sizeof(StringRef), true}, + {"FIRST_ERROR_MSG", TYPE_STRING, sizeof(StringRef), true}, }; SchemaRoutineLoadJobScanner::SchemaRoutineLoadJobScanner() @@ -170,15 +170,15 @@ Status SchemaRoutineLoadJobScanner::_fill_block_impl(Block* block) { case 16: // ERROR_LOG_URLS column_value = job_info.__isset.error_log_urls ? job_info.error_log_urls : ""; break; - case 17: // FIRST_ERROR_MSG - column_value = job_info.__isset.first_error_msg ? job_info.first_error_msg : ""; - break; - case 18: // USER_NAME + case 17: // USER_NAME column_value = job_info.__isset.user_name ? job_info.user_name : ""; break; - case 21: // COMPUTE_GROUP + case 20: // COMPUTE_GROUP column_value = job_info.__isset.compute_group ? job_info.compute_group : ""; break; + case 21: // FIRST_ERROR_MSG + column_value = job_info.__isset.first_error_msg ? job_info.first_error_msg : ""; + break; } str_refs[row_idx] = @@ -202,4 +202,4 @@ Status SchemaRoutineLoadJobScanner::_fill_block_impl(Block* block) { return Status::OK(); } -} // namespace doris \ No newline at end of file +} // namespace doris diff --git a/cloud/src/meta-service/meta_service_txn.cpp b/cloud/src/meta-service/meta_service_txn.cpp index 5c35958b1f0b83..7c12dc788830a9 100644 --- a/cloud/src/meta-service/meta_service_txn.cpp +++ b/cloud/src/meta-service/meta_service_txn.cpp @@ -578,6 +578,11 @@ void put_routine_load_progress(MetaServiceCode& code, std::string& msg, new_progress_info.mutable_partition_to_offset()->insert(elem); } } + if (commit_attachment.has_first_error_msg()) { + new_progress_info.set_first_error_msg(commit_attachment.first_error_msg()); + } else if (prev_progress_info.has_first_error_msg()) { + new_progress_info.set_first_error_msg(prev_progress_info.first_error_msg()); + } std::string new_statistic_val; RoutineLoadJobStatisticPB* new_statistic_info = new_progress_info.mutable_stat(); @@ -750,6 +755,9 @@ void MetaServiceImpl::get_rl_task_commit_attach(::google::protobuf::RpcControlle commit_attach->set_received_bytes(statistic_info.received_bytes()); commit_attach->set_task_execution_time_ms(statistic_info.task_execution_time_ms()); } + if (progress_info->has_first_error_msg()) { + commit_attach->set_first_error_msg(progress_info->first_error_msg()); + } } void MetaServiceImpl::get_streaming_task_commit_attach( diff --git a/cloud/test/meta_service_job_test.cpp b/cloud/test/meta_service_job_test.cpp index 089418745b3c89..def28eb52fefba 100644 --- a/cloud/test/meta_service_job_test.cpp +++ b/cloud/test/meta_service_job_test.cpp @@ -5334,6 +5334,71 @@ TEST(MetaServiceJobTest, IdempotentCompactionJob) { } } +TEST(MetaServiceJobTest, RoutineLoadFirstErrorMsgTest) { + auto meta_service = get_meta_service(); + ASSERT_NE(meta_service, nullptr); + + constexpr int64_t db_id = 1100; + constexpr int64_t job_id = 2100; + const std::string cloud_unique_id = "test_cloud_unique_id"; + + auto commit_routine_load_task = [&](const std::string& label, int64_t offset, + bool set_first_error_msg) { + CommitTxnRequest request; + request.set_cloud_unique_id(cloud_unique_id); + { + brpc::Controller cntl; + BeginTxnRequest begin_request; + BeginTxnResponse begin_response; + auto* txn_info = begin_request.mutable_txn_info(); + txn_info->set_db_id(db_id); + txn_info->set_label(label); + txn_info->add_table_ids(1); + txn_info->set_load_job_source_type( + LoadJobSourceTypePB::LOAD_JOB_SRC_TYPE_ROUTINE_LOAD_TASK); + txn_info->set_timeout_ms(36000); + meta_service->begin_txn(&cntl, &begin_request, &begin_response, nullptr); + ASSERT_EQ(begin_response.status().code(), MetaServiceCode::OK); + request.set_txn_id(begin_response.txn_id()); + request.set_db_id(db_id); + } + + auto* attachment = request.mutable_commit_attachment(); + attachment->set_type(TxnCommitAttachmentPB::RT_TASK_TXN_COMMIT_ATTACHMENT); + auto* routine_load_attachment = attachment->mutable_rl_task_txn_commit_attachment(); + routine_load_attachment->set_job_id(job_id); + (*routine_load_attachment->mutable_progress()->mutable_partition_to_offset())[0] = offset; + routine_load_attachment->set_loaded_rows(1); + if (set_first_error_msg) { + routine_load_attachment->set_first_error_msg("invalid source row"); + } + + CommitTxnResponse response; + brpc::Controller cntl; + meta_service->commit_txn(&cntl, &request, &response, nullptr); + EXPECT_FALSE(cntl.Failed()) << "Error: " << cntl.ErrorText(); + EXPECT_EQ(response.status().code(), MetaServiceCode::OK); + }; + + commit_routine_load_task("routine_load_first_error_msg_1", 1, true); + commit_routine_load_task("routine_load_first_error_msg_2", 2, false); + + GetRLTaskCommitAttachRequest request; + request.set_cloud_unique_id(cloud_unique_id); + request.set_db_id(db_id); + request.set_job_id(job_id); + GetRLTaskCommitAttachResponse response; + brpc::Controller cntl; + meta_service->get_rl_task_commit_attach(&cntl, &request, &response, nullptr); + + EXPECT_FALSE(cntl.Failed()) << "Error: " << cntl.ErrorText(); + ASSERT_EQ(response.status().code(), MetaServiceCode::OK); + ASSERT_TRUE(response.has_commit_attach()); + EXPECT_EQ(response.commit_attach().first_error_msg(), "invalid source row"); + EXPECT_EQ(response.commit_attach().loaded_rows(), 2); + EXPECT_EQ(response.commit_attach().progress().partition_to_offset().at(0), 2); +} + TEST(MetaServiceJobTest, UpdateStreamingJobMetaTest) { auto meta_service = get_meta_service(); ASSERT_NE(meta_service, nullptr); diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java index d70fa9ed9f98ed..bc20f860d16576 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java @@ -702,11 +702,11 @@ public class SchemaTable extends Table { .column("LAG", ScalarType.createStringType()) .column("REASON_OF_STATE_CHANGED", ScalarType.createStringType()) .column("ERROR_LOG_URLS", ScalarType.createStringType()) - .column("FIRST_ERROR_MSG", ScalarType.createStringType()) .column("USER_NAME", ScalarType.createStringType()) .column("CURRENT_ABORT_TASK_NUM", ScalarType.createType(PrimitiveType.INT)) .column("IS_ABNORMAL_PAUSE", ScalarType.createType(PrimitiveType.BOOLEAN)) .column("COMPUTE_GROUP", ScalarType.createStringType()) + .column("FIRST_ERROR_MSG", ScalarType.createStringType()) .build()) ) .put("load_jobs", 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 8bc795f32e472f..0b794cbd149356 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 @@ -907,6 +907,7 @@ protected void updateCloudProgress(RLTaskTxnCommitAttachment attachment) { this.jobStatistic.unselectedRows = attachment.getUnselectedRows(); this.jobStatistic.receivedBytes = attachment.getReceivedBytes(); this.jobStatistic.totalTaskExcutionTimeMs = System.currentTimeMillis() - createTimestamp; + updateFirstErrorMsg(attachment); } private void updateNumOfData(long numOfTotalRows, long numOfErrorRows, long unselectedRows, long receivedBytes, @@ -1204,7 +1205,9 @@ public void afterCommitted(TransactionState txnState, boolean txnOperated) throw @Override public void replayOnCommitted(TransactionState txnState) { Preconditions.checkNotNull(txnState.getTxnCommitAttachment(), txnState); - replayUpdateProgress((RLTaskTxnCommitAttachment) txnState.getTxnCommitAttachment()); + RLTaskTxnCommitAttachment attachment = (RLTaskTxnCommitAttachment) txnState.getTxnCommitAttachment(); + replayUpdateProgress(attachment); + updateJobInfoFromAttachment(attachment); this.jobStatistic.committedTaskNum++; if (LOG.isDebugEnabled()) { LOG.debug("replay on committed: {}", txnState); @@ -1379,11 +1382,13 @@ public void replayOnAborted(TransactionState txnState) { // it need check commit info before update progress // for follower FE node progress may exceed correct progress // the data will lost if FE leader change at this moment - if (txnState.getTxnCommitAttachment() != null - && checkCommitInfo((RLTaskTxnCommitAttachment) txnState.getTxnCommitAttachment(), - txnState, - TransactionState.TxnStatusChangeReason.fromString(txnState.getReason()))) { - replayUpdateProgress((RLTaskTxnCommitAttachment) txnState.getTxnCommitAttachment()); + if (txnState.getTxnCommitAttachment() != null) { + RLTaskTxnCommitAttachment attachment = (RLTaskTxnCommitAttachment) txnState.getTxnCommitAttachment(); + if (checkCommitInfo(attachment, txnState, + TransactionState.TxnStatusChangeReason.fromString(txnState.getReason()))) { + replayUpdateProgress(attachment); + } + updateJobInfoFromAttachment(attachment); } this.jobStatistic.abortedTaskNum++; if (LOG.isDebugEnabled()) { @@ -1414,14 +1419,8 @@ private void executeTaskOnTxnStatusChanged(RoutineLoadTaskInfo routineLoadTaskIn routineLoadTaskInfo.handleTaskByTxnCommitAttachment(rlTaskTxnCommitAttachment); } - if (rlTaskTxnCommitAttachment != null - && !Strings.isNullOrEmpty(rlTaskTxnCommitAttachment.getFirstErrorMsg())) { - firstErrorMsg = StringUtils.abbreviate( - rlTaskTxnCommitAttachment.getFirstErrorMsg(), Config.first_error_msg_max_length); - } - - if (rlTaskTxnCommitAttachment != null && !Strings.isNullOrEmpty(rlTaskTxnCommitAttachment.getErrorLogUrl())) { - errorLogUrls.add(rlTaskTxnCommitAttachment.getErrorLogUrl()); + if (rlTaskTxnCommitAttachment != null) { + updateJobInfoFromAttachment(rlTaskTxnCommitAttachment); } routineLoadTaskInfo.setTxnStatus(txnStatus); @@ -1438,6 +1437,20 @@ private void executeTaskOnTxnStatusChanged(RoutineLoadTaskInfo routineLoadTaskIn } } + private void updateJobInfoFromAttachment(RLTaskTxnCommitAttachment attachment) { + updateFirstErrorMsg(attachment); + if (!Strings.isNullOrEmpty(attachment.getErrorLogUrl())) { + errorLogUrls.add(attachment.getErrorLogUrl()); + } + } + + private void updateFirstErrorMsg(RLTaskTxnCommitAttachment attachment) { + if (!Strings.isNullOrEmpty(attachment.getFirstErrorMsg())) { + firstErrorMsg = StringUtils.abbreviate( + attachment.getFirstErrorMsg(), Config.first_error_msg_max_length); + } + } + protected static void checkMeta(OlapTable olapTable, RoutineLoadDesc routineLoadDesc) throws UserException { if (routineLoadDesc == null) { return; diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java index 4bedab14b7fe1e..17ad36600b3216 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java @@ -41,6 +41,7 @@ import org.apache.doris.transaction.GlobalTransactionMgrIface; import org.apache.doris.transaction.TransactionException; import org.apache.doris.transaction.TransactionState; +import org.apache.doris.transaction.TransactionStatus; import org.apache.doris.transaction.TxnStateCallbackFactory; import com.google.common.base.Strings; @@ -76,6 +77,47 @@ public void testFirstErrorMsgInTxnCommitAttachment() { Assert.assertEquals("invalid source row", cloudAttachment.getFirstErrorMsg()); } + @Test + public void testFirstErrorMsgOnReplay() { + TKafkaRLTaskProgress thriftProgress = new TKafkaRLTaskProgress(Maps.newHashMap()); + RLTaskTxnCommitAttachment attachment = new RLTaskTxnCommitAttachment(); + Deencapsulation.setField(attachment, "progress", new KafkaProgress(thriftProgress)); + Deencapsulation.setField(attachment, "errorLogUrl", "http://127.0.0.1/error_log"); + Deencapsulation.setField(attachment, "firstErrorMsg", "invalid source row"); + + TransactionState committedTxnState = Mockito.mock(TransactionState.class); + Mockito.doReturn(attachment).when(committedTxnState).getTxnCommitAttachment(); + Mockito.when(committedTxnState.getTransactionStatus()).thenReturn(TransactionStatus.COMMITTED); + RoutineLoadJob committedJob = new KafkaRoutineLoadJob(); + Deencapsulation.setField(committedJob, "progress", new KafkaProgress(thriftProgress)); + committedJob.replayOnCommitted(committedTxnState); + Assert.assertEquals("http://127.0.0.1/error_log", committedJob.getErrorLogUrls().peek()); + Assert.assertEquals("invalid source row", committedJob.getFirstErrorMsg()); + + TransactionState abortedTxnState = Mockito.mock(TransactionState.class); + Mockito.doReturn(attachment).when(abortedTxnState).getTxnCommitAttachment(); + Mockito.when(abortedTxnState.getTransactionStatus()).thenReturn(TransactionStatus.ABORTED); + Mockito.when(abortedTxnState.getReason()).thenReturn("test abort"); + RoutineLoadJob abortedJob = new KafkaRoutineLoadJob(); + abortedJob.replayOnAborted(abortedTxnState); + Assert.assertEquals("http://127.0.0.1/error_log", abortedJob.getErrorLogUrls().peek()); + Assert.assertEquals("invalid source row", abortedJob.getFirstErrorMsg()); + } + + @Test + public void testFirstErrorMsgFromCloudProgress() { + TKafkaRLTaskProgress thriftProgress = new TKafkaRLTaskProgress(Maps.newHashMap()); + RLTaskTxnCommitAttachment attachment = new RLTaskTxnCommitAttachment(); + Deencapsulation.setField(attachment, "progress", new KafkaProgress(thriftProgress)); + Deencapsulation.setField(attachment, "firstErrorMsg", "invalid source row"); + + RoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(); + Deencapsulation.setField(routineLoadJob, "progress", new KafkaProgress(thriftProgress)); + routineLoadJob.updateCloudProgress(attachment); + + Assert.assertEquals("invalid source row", routineLoadJob.getFirstErrorMsg()); + } + @Test public void testAfterAbortedReasonOffsetOutOfRange() throws UserException { Env env = Mockito.mock(Env.class); diff --git a/gensrc/proto/cloud.proto b/gensrc/proto/cloud.proto index 6c85e190c01e4e..17dcf375ff0329 100644 --- a/gensrc/proto/cloud.proto +++ b/gensrc/proto/cloud.proto @@ -394,6 +394,7 @@ message TxnCoordinatorPB { message RoutineLoadProgressPB { map partition_to_offset = 1; optional RoutineLoadJobStatisticPB stat = 2; + optional string first_error_msg = 3; } message RLTaskTxnCommitAttachmentPB { diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load_first_error_msg.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load_first_error_msg.groovy index 5fea612394daa8..24efca3b0f707e 100644 --- a/regression-test/suites/load_p0/routine_load/test_routine_load_first_error_msg.groovy +++ b/regression-test/suites/load_p0/routine_load/test_routine_load_first_error_msg.groovy @@ -29,7 +29,6 @@ suite("test_routine_load_first_error_msg", "p0") { def kafkaBroker = RoutineLoadTestUtils.getKafkaBroker(context) def kafkaTopic = "test_routine_load_first_error_msg_${System.currentTimeMillis()}" def jobName = "test_routine_load_first_error_msg" - def tableName = "test_routine_load_first_error_msg" def adminProps = new Properties() adminProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBroker) @@ -41,9 +40,9 @@ suite("test_routine_load_first_error_msg", "p0") { } try { - sql "DROP TABLE IF EXISTS ${tableName}" + sql "DROP TABLE IF EXISTS test_routine_load_first_error_msg" sql """ - CREATE TABLE ${tableName} ( + CREATE TABLE test_routine_load_first_error_msg ( id INT, name STRING ) @@ -55,7 +54,7 @@ suite("test_routine_load_first_error_msg", "p0") { """ sql """ - CREATE ROUTINE LOAD ${jobName} ON ${tableName} + CREATE ROUTINE LOAD ${jobName} ON test_routine_load_first_error_msg COLUMNS TERMINATED BY "|" PROPERTIES ( "max_error_number" = "10", @@ -98,7 +97,7 @@ suite("test_routine_load_first_error_msg", "p0") { WHERE JOB_NAME = '${jobName}' """ def showResult = sql "SHOW ROUTINE LOAD FOR ${jobName}" - def loadedRows = sql "SELECT COUNT(*) FROM ${tableName}" + def loadedRows = sql "SELECT COUNT(*) FROM test_routine_load_first_error_msg" def informationSchemaErrorLogUrls = jobInfo[0][0].toString() def firstErrorMsg = jobInfo[0][1] def showErrorLogUrls = showResult[0][18].toString() @@ -124,6 +123,5 @@ suite("test_routine_load_first_error_msg", "p0") { } } finally { try_sql "STOP ROUTINE LOAD FOR ${jobName}" - try_sql "DROP TABLE IF EXISTS ${tableName}" } } From b415c12fa85a764caa03dc23d6a8454126bca268 Mon Sep 17 00:00:00 2001 From: laihui Date: Tue, 14 Jul 2026 09:22:43 +0800 Subject: [PATCH 03/11] [fix](load) Bound routine load first error messages Cause: Routine load attachments retained the full first error message before durable storage, and the regression case assumed that the message always contained the raw source line. Solution: Abbreviate first error messages when constructing routine load transaction attachments, cover overlong messages across cloud serialization, and validate the regression result using a stable bad-row marker. --- .../load/routineload/RLTaskTxnCommitAttachment.java | 10 ++++++++-- .../doris/load/routineload/RoutineLoadJobTest.java | 9 ++++++--- .../test_routine_load_first_error_msg.groovy | 4 ++-- 3 files changed, 16 insertions(+), 7 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RLTaskTxnCommitAttachment.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RLTaskTxnCommitAttachment.java index 71d85c264ec447..48607266075ff1 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RLTaskTxnCommitAttachment.java +++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RLTaskTxnCommitAttachment.java @@ -18,6 +18,7 @@ package org.apache.doris.load.routineload; import org.apache.doris.cloud.proto.Cloud.RLTaskTxnCommitAttachmentPB; +import org.apache.doris.common.Config; import org.apache.doris.load.routineload.kafka.KafkaProgress; import org.apache.doris.load.routineload.kinesis.KinesisProgress; import org.apache.doris.thrift.TRLTaskTxnCommitAttachment; @@ -26,6 +27,7 @@ import org.apache.doris.transaction.TxnCommitAttachment; import com.google.gson.annotations.SerializedName; +import org.apache.commons.lang3.StringUtils; // {"progress": "", "backendId": "", "taskSignature": "", "numOfErrorData": "", // "numOfTotalData": "", "taskId": "", "jobId": ""} @@ -77,7 +79,7 @@ public RLTaskTxnCommitAttachment(TRLTaskTxnCommitAttachment rlTaskTxnCommitAttac this.errorLogUrl = rlTaskTxnCommitAttachment.getErrorLogUrl(); } if (rlTaskTxnCommitAttachment.isSetFirstErrorMsg()) { - this.firstErrorMsg = rlTaskTxnCommitAttachment.getFirstErrorMsg(); + this.firstErrorMsg = abbreviateFirstErrorMsg(rlTaskTxnCommitAttachment.getFirstErrorMsg()); } } @@ -96,7 +98,11 @@ public RLTaskTxnCommitAttachment(RLTaskTxnCommitAttachmentPB rlTaskTxnCommitAtta this.progress = progress; this.errorLogUrl = rlTaskTxnCommitAttachment.getErrorLogUrl(); - this.firstErrorMsg = rlTaskTxnCommitAttachment.getFirstErrorMsg(); + this.firstErrorMsg = abbreviateFirstErrorMsg(rlTaskTxnCommitAttachment.getFirstErrorMsg()); + } + + private static String abbreviateFirstErrorMsg(String firstErrorMsg) { + return StringUtils.abbreviate(firstErrorMsg, Config.first_error_msg_max_length); } public long getJobId() { diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java index 17ad36600b3216..b7752dc8b73d55 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java @@ -22,6 +22,7 @@ import org.apache.doris.catalog.Env; import org.apache.doris.catalog.Table; import org.apache.doris.cloud.transaction.TxnUtil; +import org.apache.doris.common.Config; import org.apache.doris.common.InternalErrorCode; import org.apache.doris.common.Pair; import org.apache.doris.common.UserException; @@ -60,21 +61,23 @@ public class RoutineLoadJobTest { @Test public void testFirstErrorMsgInTxnCommitAttachment() { + String overlongFirstErrorMsg = Strings.repeat("x", Config.first_error_msg_max_length + 1); TRLTaskTxnCommitAttachment thriftAttachment = new TRLTaskTxnCommitAttachment(); thriftAttachment.setLoadSourceType(TLoadSourceType.KAFKA); thriftAttachment.setId(new TUniqueId(1, 2)); thriftAttachment.setJobId(3); thriftAttachment.setKafkaRLTaskProgress(new TKafkaRLTaskProgress(Maps.newHashMap())); thriftAttachment.setErrorLogUrl("http://127.0.0.1/error_log"); - thriftAttachment.setFirstErrorMsg("invalid source row"); + thriftAttachment.setFirstErrorMsg(overlongFirstErrorMsg); RLTaskTxnCommitAttachment attachment = new RLTaskTxnCommitAttachment(thriftAttachment); - Assert.assertEquals("invalid source row", attachment.getFirstErrorMsg()); + Assert.assertEquals(Config.first_error_msg_max_length, attachment.getFirstErrorMsg().length()); + Assert.assertTrue(attachment.getFirstErrorMsg().endsWith("...")); RLTaskTxnCommitAttachment cloudAttachment = TxnUtil.rtTaskTxnCommitAttachmentFromPb( TxnUtil.rlTaskTxnCommitAttachmentToPb(attachment)); Assert.assertEquals("http://127.0.0.1/error_log", cloudAttachment.getErrorLogUrl()); - Assert.assertEquals("invalid source row", cloudAttachment.getFirstErrorMsg()); + Assert.assertEquals(attachment.getFirstErrorMsg(), cloudAttachment.getFirstErrorMsg()); } @Test diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load_first_error_msg.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load_first_error_msg.groovy index 24efca3b0f707e..a41c84d2ce04a0 100644 --- a/regression-test/suites/load_p0/routine_load/test_routine_load_first_error_msg.groovy +++ b/regression-test/suites/load_p0/routine_load/test_routine_load_first_error_msg.groovy @@ -103,9 +103,9 @@ suite("test_routine_load_first_error_msg", "p0") { def showErrorLogUrls = showResult[0][18].toString() def showFirstErrorMsg = showResult[0][23] if (loadedRows[0][0] == 1 && firstErrorMsg != null - && firstErrorMsg.toString().contains("100|bad_row") + && firstErrorMsg.toString().contains("bad_row") && showFirstErrorMsg != null - && showFirstErrorMsg.toString().contains("100|bad_row")) { + && showFirstErrorMsg.toString().contains("bad_row")) { assertEquals(informationSchemaErrorLogUrls, showErrorLogUrls) assertFalse(informationSchemaErrorLogUrls.contains("first_error_msg:")) assertFalse(showErrorLogUrls.contains("first_error_msg:")) From 76b0771e64bf7d86a755872f5c937d417022a434 Mon Sep 17 00:00:00 2001 From: laihui Date: Tue, 14 Jul 2026 10:38:58 +0800 Subject: [PATCH 04/11] [fix](load) Preserve routine load errors across cloud paths Cause: Cloud abort and progress reset paths dropped routine load first error messages, while the successful ordinary stream load path was unintentionally changed to expose the raw message. Solution: Forward routine load attachments through cloud abort, preserve first error messages when resetting cloud progress, keep successful ordinary stream load behavior unchanged, and add focused cloud abort and reset coverage. --- .../load/stream_load/stream_load_executor.cpp | 4 +- cloud/src/meta-service/meta_service_txn.cpp | 3 ++ cloud/test/meta_service_job_test.cpp | 21 ++++++++ .../CloudGlobalTransactionMgr.java | 6 ++- .../CloudGlobalTransactionMgrTest.java | 50 +++++++++++++++++++ 5 files changed, 82 insertions(+), 2 deletions(-) diff --git a/be/src/load/stream_load/stream_load_executor.cpp b/be/src/load/stream_load/stream_load_executor.cpp index 66c1d7fd339f30..9b307c932d4776 100644 --- a/be/src/load/stream_load/stream_load_executor.cpp +++ b/be/src/load/stream_load/stream_load_executor.cpp @@ -99,7 +99,6 @@ Status StreamLoadExecutor::execute_plan_fragment( ctx->loaded_bytes = state->num_bytes_load_total(); int64_t num_selected_rows = ctx->number_total_rows - ctx->number_unselected_rows; ctx->error_url = to_load_error_http_path(state->get_error_log_file_path()); - ctx->first_error_msg = state->get_first_error_msg(); if (status->ok() && !ctx->group_commit && num_selected_rows > 0 && (double)ctx->number_filtered_rows / num_selected_rows > ctx->max_filter_ratio) { // NOTE: Do not modify the error message here, for historical reasons, @@ -111,6 +110,9 @@ Status StreamLoadExecutor::execute_plan_fragment( *status = Status::DataQualityError("too many filtered rows"); } } + if (ctx->load_type == TLoadType::ROUTINE_LOAD || !status->ok()) { + ctx->first_error_msg = state->get_first_error_msg(); + } if (status->ok()) { DorisMetrics::instance()->stream_receive_bytes_total->increment(ctx->receive_bytes); diff --git a/cloud/src/meta-service/meta_service_txn.cpp b/cloud/src/meta-service/meta_service_txn.cpp index 7c12dc788830a9..436544c22db144 100644 --- a/cloud/src/meta-service/meta_service_txn.cpp +++ b/cloud/src/meta-service/meta_service_txn.cpp @@ -895,6 +895,9 @@ void MetaServiceImpl::reset_rl_progress(::google::protobuf::RpcController* contr } } } + if (prev_progress_info.has_first_error_msg()) { + new_progress_info.set_first_error_msg(prev_progress_info.first_error_msg()); + } if (!new_progress_info.SerializeToString(&new_progress_val)) { code = MetaServiceCode::PROTOBUF_SERIALIZE_ERR; diff --git a/cloud/test/meta_service_job_test.cpp b/cloud/test/meta_service_job_test.cpp index def28eb52fefba..c214b037fa0f5d 100644 --- a/cloud/test/meta_service_job_test.cpp +++ b/cloud/test/meta_service_job_test.cpp @@ -5397,6 +5397,27 @@ TEST(MetaServiceJobTest, RoutineLoadFirstErrorMsgTest) { EXPECT_EQ(response.commit_attach().first_error_msg(), "invalid source row"); EXPECT_EQ(response.commit_attach().loaded_rows(), 2); EXPECT_EQ(response.commit_attach().progress().partition_to_offset().at(0), 2); + + ResetRLProgressRequest reset_request; + reset_request.set_cloud_unique_id(cloud_unique_id); + reset_request.set_db_id(db_id); + reset_request.set_job_id(job_id); + (*reset_request.mutable_partition_to_offset())[0] = 10; + ResetRLProgressResponse reset_response; + brpc::Controller reset_cntl; + meta_service->reset_rl_progress(&reset_cntl, &reset_request, &reset_response, nullptr); + EXPECT_FALSE(reset_cntl.Failed()) << "Error: " << reset_cntl.ErrorText(); + ASSERT_EQ(reset_response.status().code(), MetaServiceCode::OK); + + GetRLTaskCommitAttachResponse response_after_reset; + brpc::Controller cntl_after_reset; + meta_service->get_rl_task_commit_attach(&cntl_after_reset, &request, &response_after_reset, + nullptr); + EXPECT_FALSE(cntl_after_reset.Failed()) << "Error: " << cntl_after_reset.ErrorText(); + ASSERT_EQ(response_after_reset.status().code(), MetaServiceCode::OK); + ASSERT_TRUE(response_after_reset.has_commit_attach()); + EXPECT_EQ(response_after_reset.commit_attach().first_error_msg(), "invalid source row"); + EXPECT_EQ(response_after_reset.commit_attach().progress().partition_to_offset().at(0), 10); } TEST(MetaServiceJobTest, UpdateStreamingJobMetaTest) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java index 0749d38f3111c4..1cf69114d0981e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java @@ -1841,7 +1841,7 @@ public void abortTransaction(Long dbId, Long transactionId, String reason, AbortTxnResponse abortTxnResponse = null; try { - abortTxnResponse = abortTransactionImpl(dbId, transactionId, reason, null); + abortTxnResponse = abortTransactionImpl(dbId, transactionId, reason, txnCommitAttachment); } finally { handleAfterAbort(abortTxnResponse, txnCommitAttachment, transactionId, callbackInfo.first, callbackInfo.second); @@ -1905,6 +1905,10 @@ private AbortTxnResponse abortTransactionImpl(Long dbId, Long transactionId, Str builder.setTxnId(transactionId); builder.setReason(reason); builder.setCloudUniqueId(Config.cloud_unique_id); + if (txnCommitAttachment instanceof RLTaskTxnCommitAttachment) { + builder.setCommitAttachment(TxnUtil.rlTaskTxnCommitAttachmentToPb( + (RLTaskTxnCommitAttachment) txnCommitAttachment)); + } final AbortTxnRequest abortTxnRequest = builder.build(); AbortTxnResponse abortTxnResponse = null; diff --git a/fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgrTest.java b/fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgrTest.java index bcc5b41c3e8465..a6db5409e897f3 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgrTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgrTest.java @@ -36,8 +36,10 @@ import org.apache.doris.common.FeMetaVersion; import org.apache.doris.common.LabelAlreadyUsedException; import org.apache.doris.common.UserException; +import org.apache.doris.load.routineload.RLTaskTxnCommitAttachment; import org.apache.doris.transaction.GlobalTransactionMgrIface; import org.apache.doris.transaction.TransactionState; +import org.apache.doris.transaction.TxnStateChangeCallback; import com.google.common.collect.Lists; import org.junit.After; @@ -45,6 +47,7 @@ import org.junit.Before; import org.junit.Test; import org.junit.jupiter.api.Assertions; +import org.mockito.ArgumentCaptor; import org.mockito.MockedStatic; import org.mockito.Mockito; @@ -306,6 +309,53 @@ public void testAbortTransaction() throws Exception { } } + @Test + public void testAbortRoutineLoadTransactionWithAttachment() throws Exception { + long transactionId = 123534; + long jobId = 1001; + RLTaskTxnCommitAttachment attachment = new RLTaskTxnCommitAttachment( + Cloud.RLTaskTxnCommitAttachmentPB.newBuilder() + .setJobId(jobId) + .setTaskId(Cloud.UniqueIdPB.newBuilder().setHi(1).setLo(2)) + .setProgress(Cloud.RoutineLoadProgressPB.newBuilder()) + .setFirstErrorMsg("invalid source row") + .build()); + TxnStateChangeCallback callback = Mockito.mock(TxnStateChangeCallback.class); + Mockito.when(callback.getId()).thenReturn(jobId); + masterTransMgr.getCallbackFactory().addCallback(callback); + + MetaServiceProxy mockProxy = Mockito.mock(MetaServiceProxy.class); + try (MockedStatic mockedStatic = Mockito.mockStatic(MetaServiceProxy.class)) { + mockedStatic.when(MetaServiceProxy::getInstance).thenReturn(mockProxy); + Mockito.doAnswer(invocation -> { + Cloud.AbortTxnRequest request = invocation.getArgument(0); + Assert.assertTrue(request.hasCommitAttachment()); + Assert.assertEquals("invalid source row", request.getCommitAttachment() + .getRlTaskTxnCommitAttachment().getFirstErrorMsg()); + return AbortTxnResponse.newBuilder() + .setStatus(Cloud.MetaServiceResponseStatus.newBuilder() + .setCode(MetaServiceCode.OK).setMsg("OK")) + .setTxnInfo(buildTxnInfo(transactionId).toBuilder() + .setStatus(Cloud.TxnStatusPB.TXN_STATUS_ABORTED) + .setReason("data quality error") + .setCommitAttachment(request.getCommitAttachment())) + .build(); + }).when(mockProxy).abortTxn(Mockito.any()); + + masterTransMgr.abortTransaction(CatalogTestUtil.testDbId1, transactionId, + "data quality error", attachment, Lists.newArrayList()); + + ArgumentCaptor txnStateCaptor = ArgumentCaptor.forClass(TransactionState.class); + Mockito.verify(callback).afterAborted(txnStateCaptor.capture(), Mockito.eq(true), + Mockito.eq("data quality error")); + RLTaskTxnCommitAttachment callbackAttachment = + (RLTaskTxnCommitAttachment) txnStateCaptor.getValue().getTxnCommitAttachment(); + Assert.assertEquals("invalid source row", callbackAttachment.getFirstErrorMsg()); + } finally { + masterTransMgr.getCallbackFactory().removeCallback(jobId); + } + } + @Test public void testAbortTransactionByLabel() throws Exception { MetaServiceProxy mockProxy = Mockito.mock(MetaServiceProxy.class); From 14cd1e1b94d1e0026e25e3f75dd1d5f7b6bbbc4b Mon Sep 17 00:00:00 2001 From: laihui Date: Tue, 14 Jul 2026 10:55:52 +0800 Subject: [PATCH 05/11] [fix](load) Expire routine load error diagnostics together Cause: Routine load error URLs point to temporary BE files, but the first error message was persisted independently in cloud progress and job metadata, allowing stale diagnostics to outlive their URLs. Solution: Keep error URLs and first error messages transient, expire them together using the error-log retention period, stop persisting the message in cloud routine load progress, and update the focused coverage. --- cloud/src/meta-service/meta_service_txn.cpp | 13 --- cloud/test/meta_service_job_test.cpp | 86 ------------------- .../java/org/apache/doris/common/Config.java | 5 ++ .../load/routineload/RoutineLoadJob.java | 35 +++++--- .../load/routineload/RoutineLoadJobTest.java | 24 +++--- gensrc/proto/cloud.proto | 2 +- 6 files changed, 41 insertions(+), 124 deletions(-) diff --git a/cloud/src/meta-service/meta_service_txn.cpp b/cloud/src/meta-service/meta_service_txn.cpp index 436544c22db144..c5a475a4f4aea6 100644 --- a/cloud/src/meta-service/meta_service_txn.cpp +++ b/cloud/src/meta-service/meta_service_txn.cpp @@ -578,12 +578,6 @@ void put_routine_load_progress(MetaServiceCode& code, std::string& msg, new_progress_info.mutable_partition_to_offset()->insert(elem); } } - if (commit_attachment.has_first_error_msg()) { - new_progress_info.set_first_error_msg(commit_attachment.first_error_msg()); - } else if (prev_progress_info.has_first_error_msg()) { - new_progress_info.set_first_error_msg(prev_progress_info.first_error_msg()); - } - std::string new_statistic_val; RoutineLoadJobStatisticPB* new_statistic_info = new_progress_info.mutable_stat(); if (prev_progress_info.has_stat()) { @@ -755,9 +749,6 @@ void MetaServiceImpl::get_rl_task_commit_attach(::google::protobuf::RpcControlle commit_attach->set_received_bytes(statistic_info.received_bytes()); commit_attach->set_task_execution_time_ms(statistic_info.task_execution_time_ms()); } - if (progress_info->has_first_error_msg()) { - commit_attach->set_first_error_msg(progress_info->first_error_msg()); - } } void MetaServiceImpl::get_streaming_task_commit_attach( @@ -895,10 +886,6 @@ void MetaServiceImpl::reset_rl_progress(::google::protobuf::RpcController* contr } } } - if (prev_progress_info.has_first_error_msg()) { - new_progress_info.set_first_error_msg(prev_progress_info.first_error_msg()); - } - if (!new_progress_info.SerializeToString(&new_progress_val)) { code = MetaServiceCode::PROTOBUF_SERIALIZE_ERR; ss << "failed to serialize new progress val" diff --git a/cloud/test/meta_service_job_test.cpp b/cloud/test/meta_service_job_test.cpp index c214b037fa0f5d..089418745b3c89 100644 --- a/cloud/test/meta_service_job_test.cpp +++ b/cloud/test/meta_service_job_test.cpp @@ -5334,92 +5334,6 @@ TEST(MetaServiceJobTest, IdempotentCompactionJob) { } } -TEST(MetaServiceJobTest, RoutineLoadFirstErrorMsgTest) { - auto meta_service = get_meta_service(); - ASSERT_NE(meta_service, nullptr); - - constexpr int64_t db_id = 1100; - constexpr int64_t job_id = 2100; - const std::string cloud_unique_id = "test_cloud_unique_id"; - - auto commit_routine_load_task = [&](const std::string& label, int64_t offset, - bool set_first_error_msg) { - CommitTxnRequest request; - request.set_cloud_unique_id(cloud_unique_id); - { - brpc::Controller cntl; - BeginTxnRequest begin_request; - BeginTxnResponse begin_response; - auto* txn_info = begin_request.mutable_txn_info(); - txn_info->set_db_id(db_id); - txn_info->set_label(label); - txn_info->add_table_ids(1); - txn_info->set_load_job_source_type( - LoadJobSourceTypePB::LOAD_JOB_SRC_TYPE_ROUTINE_LOAD_TASK); - txn_info->set_timeout_ms(36000); - meta_service->begin_txn(&cntl, &begin_request, &begin_response, nullptr); - ASSERT_EQ(begin_response.status().code(), MetaServiceCode::OK); - request.set_txn_id(begin_response.txn_id()); - request.set_db_id(db_id); - } - - auto* attachment = request.mutable_commit_attachment(); - attachment->set_type(TxnCommitAttachmentPB::RT_TASK_TXN_COMMIT_ATTACHMENT); - auto* routine_load_attachment = attachment->mutable_rl_task_txn_commit_attachment(); - routine_load_attachment->set_job_id(job_id); - (*routine_load_attachment->mutable_progress()->mutable_partition_to_offset())[0] = offset; - routine_load_attachment->set_loaded_rows(1); - if (set_first_error_msg) { - routine_load_attachment->set_first_error_msg("invalid source row"); - } - - CommitTxnResponse response; - brpc::Controller cntl; - meta_service->commit_txn(&cntl, &request, &response, nullptr); - EXPECT_FALSE(cntl.Failed()) << "Error: " << cntl.ErrorText(); - EXPECT_EQ(response.status().code(), MetaServiceCode::OK); - }; - - commit_routine_load_task("routine_load_first_error_msg_1", 1, true); - commit_routine_load_task("routine_load_first_error_msg_2", 2, false); - - GetRLTaskCommitAttachRequest request; - request.set_cloud_unique_id(cloud_unique_id); - request.set_db_id(db_id); - request.set_job_id(job_id); - GetRLTaskCommitAttachResponse response; - brpc::Controller cntl; - meta_service->get_rl_task_commit_attach(&cntl, &request, &response, nullptr); - - EXPECT_FALSE(cntl.Failed()) << "Error: " << cntl.ErrorText(); - ASSERT_EQ(response.status().code(), MetaServiceCode::OK); - ASSERT_TRUE(response.has_commit_attach()); - EXPECT_EQ(response.commit_attach().first_error_msg(), "invalid source row"); - EXPECT_EQ(response.commit_attach().loaded_rows(), 2); - EXPECT_EQ(response.commit_attach().progress().partition_to_offset().at(0), 2); - - ResetRLProgressRequest reset_request; - reset_request.set_cloud_unique_id(cloud_unique_id); - reset_request.set_db_id(db_id); - reset_request.set_job_id(job_id); - (*reset_request.mutable_partition_to_offset())[0] = 10; - ResetRLProgressResponse reset_response; - brpc::Controller reset_cntl; - meta_service->reset_rl_progress(&reset_cntl, &reset_request, &reset_response, nullptr); - EXPECT_FALSE(reset_cntl.Failed()) << "Error: " << reset_cntl.ErrorText(); - ASSERT_EQ(reset_response.status().code(), MetaServiceCode::OK); - - GetRLTaskCommitAttachResponse response_after_reset; - brpc::Controller cntl_after_reset; - meta_service->get_rl_task_commit_attach(&cntl_after_reset, &request, &response_after_reset, - nullptr); - EXPECT_FALSE(cntl_after_reset.Failed()) << "Error: " << cntl_after_reset.ErrorText(); - ASSERT_EQ(response_after_reset.status().code(), MetaServiceCode::OK); - ASSERT_TRUE(response_after_reset.has_commit_attach()); - EXPECT_EQ(response_after_reset.commit_attach().first_error_msg(), "invalid source row"); - EXPECT_EQ(response_after_reset.commit_attach().progress().partition_to_offset().at(0), 10); -} - TEST(MetaServiceJobTest, UpdateStreamingJobMetaTest) { auto meta_service = get_meta_service(); ASSERT_NE(meta_service, nullptr); diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java index 00526509177152..7230bae94e833f 100644 --- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java +++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java @@ -3638,6 +3638,11 @@ public static int metaServiceRpcRetryTimes() { "The maximum length of the first row error message when data quality error occurs, default is 256 bytes"}) public static int first_error_msg_max_length = 256; + @ConfField(mutable = true, description = { + "Retention time for routine load error log URLs and first error messages, in hours. " + + "It should be consistent with the BE load_error_log_reserve_hours setting."}) + public static long load_error_log_reserve_hours = 48; + @ConfField(mutable = false, description = { "Whether to enable file cache admission control(Blocklist and Allowlist)" }) 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 0b794cbd149356..fbbc92c8c7ab15 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 @@ -103,6 +103,7 @@ import java.util.Optional; import java.util.Queue; import java.util.UUID; +import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.ReentrantReadWriteLock; /** @@ -274,10 +275,11 @@ public boolean isFinalState() { protected Expr deleteCondition; // TODO(ml): error sample - // save the latest 3 error log urls - private Queue errorLogUrls = EvictingQueue.create(3); - @SerializedName("fem") - private String firstErrorMsg = ""; + // Save the latest 3 error log URLs in memory. BE error logs expire, so these diagnostics + // and their corresponding first error message should not be persisted with the job. + private transient Queue errorLogUrls = EvictingQueue.create(3); + private transient Queue errorLogUrlTimestamps = EvictingQueue.create(3); + private transient String firstErrorMsg = ""; @SerializedName("ccid") private String cloudClusterId; @@ -819,6 +821,7 @@ public String getFirstErrorMsg() { public void processTimeoutTasks() { writeLock(); try { + unprotectCleanExpiredErrorLogUrls(); List runningTasks = new ArrayList<>(routineLoadTaskInfoList); for (RoutineLoadTaskInfo routineLoadTaskInfo : runningTasks) { if (routineLoadTaskInfo.isTimeout()) { @@ -907,7 +910,6 @@ protected void updateCloudProgress(RLTaskTxnCommitAttachment attachment) { this.jobStatistic.unselectedRows = attachment.getUnselectedRows(); this.jobStatistic.receivedBytes = attachment.getReceivedBytes(); this.jobStatistic.totalTaskExcutionTimeMs = System.currentTimeMillis() - createTimestamp; - updateFirstErrorMsg(attachment); } private void updateNumOfData(long numOfTotalRows, long numOfErrorRows, long unselectedRows, long receivedBytes, @@ -1438,16 +1440,26 @@ private void executeTaskOnTxnStatusChanged(RoutineLoadTaskInfo routineLoadTaskIn } private void updateJobInfoFromAttachment(RLTaskTxnCommitAttachment attachment) { - updateFirstErrorMsg(attachment); if (!Strings.isNullOrEmpty(attachment.getErrorLogUrl())) { errorLogUrls.add(attachment.getErrorLogUrl()); + errorLogUrlTimestamps.add(System.currentTimeMillis()); + firstErrorMsg = StringUtils.abbreviate( + Strings.nullToEmpty(attachment.getFirstErrorMsg()), Config.first_error_msg_max_length); } } - private void updateFirstErrorMsg(RLTaskTxnCommitAttachment attachment) { - if (!Strings.isNullOrEmpty(attachment.getFirstErrorMsg())) { - firstErrorMsg = StringUtils.abbreviate( - attachment.getFirstErrorMsg(), Config.first_error_msg_max_length); + private void unprotectCleanExpiredErrorLogUrls() { + if (Config.load_error_log_reserve_hours < 0) { + return; + } + long expireBefore = System.currentTimeMillis() + - TimeUnit.HOURS.toMillis(Config.load_error_log_reserve_hours); + while (!errorLogUrlTimestamps.isEmpty() && errorLogUrlTimestamps.peek() <= expireBefore) { + errorLogUrlTimestamps.poll(); + errorLogUrls.poll(); + } + if (errorLogUrls.isEmpty()) { + firstErrorMsg = ""; } } @@ -1977,6 +1989,9 @@ public void write(DataOutput out) throws IOException { @Override public void gsonPostProcess() throws IOException { + errorLogUrls = EvictingQueue.create(3); + errorLogUrlTimestamps = EvictingQueue.create(3); + firstErrorMsg = ""; if (tableId == 0) { isMultiTable = true; } diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java index b7752dc8b73d55..7329a130a3ca1f 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java @@ -97,6 +97,16 @@ public void testFirstErrorMsgOnReplay() { Assert.assertEquals("http://127.0.0.1/error_log", committedJob.getErrorLogUrls().peek()); Assert.assertEquals("invalid source row", committedJob.getFirstErrorMsg()); + long originalReserveHours = Config.load_error_log_reserve_hours; + try { + Config.load_error_log_reserve_hours = 0; + committedJob.processTimeoutTasks(); + Assert.assertTrue(committedJob.getErrorLogUrls().isEmpty()); + Assert.assertEquals("", committedJob.getFirstErrorMsg()); + } finally { + Config.load_error_log_reserve_hours = originalReserveHours; + } + TransactionState abortedTxnState = Mockito.mock(TransactionState.class); Mockito.doReturn(attachment).when(abortedTxnState).getTxnCommitAttachment(); Mockito.when(abortedTxnState.getTransactionStatus()).thenReturn(TransactionStatus.ABORTED); @@ -107,20 +117,6 @@ public void testFirstErrorMsgOnReplay() { Assert.assertEquals("invalid source row", abortedJob.getFirstErrorMsg()); } - @Test - public void testFirstErrorMsgFromCloudProgress() { - TKafkaRLTaskProgress thriftProgress = new TKafkaRLTaskProgress(Maps.newHashMap()); - RLTaskTxnCommitAttachment attachment = new RLTaskTxnCommitAttachment(); - Deencapsulation.setField(attachment, "progress", new KafkaProgress(thriftProgress)); - Deencapsulation.setField(attachment, "firstErrorMsg", "invalid source row"); - - RoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(); - Deencapsulation.setField(routineLoadJob, "progress", new KafkaProgress(thriftProgress)); - routineLoadJob.updateCloudProgress(attachment); - - Assert.assertEquals("invalid source row", routineLoadJob.getFirstErrorMsg()); - } - @Test public void testAfterAbortedReasonOffsetOutOfRange() throws UserException { Env env = Mockito.mock(Env.class); diff --git a/gensrc/proto/cloud.proto b/gensrc/proto/cloud.proto index 17dcf375ff0329..e5c458132a94e1 100644 --- a/gensrc/proto/cloud.proto +++ b/gensrc/proto/cloud.proto @@ -394,7 +394,7 @@ message TxnCoordinatorPB { message RoutineLoadProgressPB { map partition_to_offset = 1; optional RoutineLoadJobStatisticPB stat = 2; - optional string first_error_msg = 3; + // Error log URLs and first error messages are transient diagnostics and are not stored here. } message RLTaskTxnCommitAttachmentPB { From f3169e03ef25f0fa690ccc3e0ac942286cd3fdd2 Mon Sep 17 00:00:00 2001 From: laihui Date: Tue, 14 Jul 2026 11:03:27 +0800 Subject: [PATCH 06/11] [fix](load) Clear routine load error diagnostics together Cause: The previous change introduced a separate retention configuration even though routine load transaction callbacks already provide the boundary for refreshing transient error diagnostics. Solution: Remove the retention configuration and timestamp tracking, clear the first error message whenever after-commit or after-abort processing clears missing error URLs, update the replay coverage, and remove an unrelated end-of-file diff. --- .../schema_routine_load_job_scanner.cpp | 2 +- .../java/org/apache/doris/common/Config.java | 5 --- .../load/routineload/RoutineLoadJob.java | 33 +++++-------------- .../load/routineload/RoutineLoadJobTest.java | 16 ++++----- 4 files changed, 16 insertions(+), 40 deletions(-) diff --git a/be/src/information_schema/schema_routine_load_job_scanner.cpp b/be/src/information_schema/schema_routine_load_job_scanner.cpp index d2d54b60911e28..0e954de76466bf 100644 --- a/be/src/information_schema/schema_routine_load_job_scanner.cpp +++ b/be/src/information_schema/schema_routine_load_job_scanner.cpp @@ -202,4 +202,4 @@ Status SchemaRoutineLoadJobScanner::_fill_block_impl(Block* block) { return Status::OK(); } -} // namespace doris +} // namespace doris \ No newline at end of file diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java index 7230bae94e833f..00526509177152 100644 --- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java +++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java @@ -3638,11 +3638,6 @@ public static int metaServiceRpcRetryTimes() { "The maximum length of the first row error message when data quality error occurs, default is 256 bytes"}) public static int first_error_msg_max_length = 256; - @ConfField(mutable = true, description = { - "Retention time for routine load error log URLs and first error messages, in hours. " - + "It should be consistent with the BE load_error_log_reserve_hours setting."}) - public static long load_error_log_reserve_hours = 48; - @ConfField(mutable = false, description = { "Whether to enable file cache admission control(Blocklist and Allowlist)" }) 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 fbbc92c8c7ab15..01e7eeda8525db 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 @@ -103,7 +103,6 @@ import java.util.Optional; import java.util.Queue; import java.util.UUID; -import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.ReentrantReadWriteLock; /** @@ -275,10 +274,9 @@ public boolean isFinalState() { protected Expr deleteCondition; // TODO(ml): error sample - // Save the latest 3 error log URLs in memory. BE error logs expire, so these diagnostics - // and their corresponding first error message should not be persisted with the job. + // Save the latest 3 error log URLs in memory. The corresponding first error message + // uses the same lifecycle and should not be persisted with the job. private transient Queue errorLogUrls = EvictingQueue.create(3); - private transient Queue errorLogUrlTimestamps = EvictingQueue.create(3); private transient String firstErrorMsg = ""; @SerializedName("ccid") @@ -821,7 +819,6 @@ public String getFirstErrorMsg() { public void processTimeoutTasks() { writeLock(); try { - unprotectCleanExpiredErrorLogUrls(); List runningTasks = new ArrayList<>(routineLoadTaskInfoList); for (RoutineLoadTaskInfo routineLoadTaskInfo : runningTasks) { if (routineLoadTaskInfo.isTimeout()) { @@ -1440,27 +1437,14 @@ private void executeTaskOnTxnStatusChanged(RoutineLoadTaskInfo routineLoadTaskIn } private void updateJobInfoFromAttachment(RLTaskTxnCommitAttachment attachment) { - if (!Strings.isNullOrEmpty(attachment.getErrorLogUrl())) { - errorLogUrls.add(attachment.getErrorLogUrl()); - errorLogUrlTimestamps.add(System.currentTimeMillis()); - firstErrorMsg = StringUtils.abbreviate( - Strings.nullToEmpty(attachment.getFirstErrorMsg()), Config.first_error_msg_max_length); - } - } - - private void unprotectCleanExpiredErrorLogUrls() { - if (Config.load_error_log_reserve_hours < 0) { - return; - } - long expireBefore = System.currentTimeMillis() - - TimeUnit.HOURS.toMillis(Config.load_error_log_reserve_hours); - while (!errorLogUrlTimestamps.isEmpty() && errorLogUrlTimestamps.peek() <= expireBefore) { - errorLogUrlTimestamps.poll(); - errorLogUrls.poll(); - } - if (errorLogUrls.isEmpty()) { + if (Strings.isNullOrEmpty(attachment.getErrorLogUrl())) { + errorLogUrls.clear(); firstErrorMsg = ""; + return; } + errorLogUrls.add(attachment.getErrorLogUrl()); + firstErrorMsg = StringUtils.abbreviate( + Strings.nullToEmpty(attachment.getFirstErrorMsg()), Config.first_error_msg_max_length); } protected static void checkMeta(OlapTable olapTable, RoutineLoadDesc routineLoadDesc) throws UserException { @@ -1990,7 +1974,6 @@ public void write(DataOutput out) throws IOException { @Override public void gsonPostProcess() throws IOException { errorLogUrls = EvictingQueue.create(3); - errorLogUrlTimestamps = EvictingQueue.create(3); firstErrorMsg = ""; if (tableId == 0) { isMultiTable = true; diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java index 7329a130a3ca1f..fea839ae018bc5 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java @@ -97,15 +97,13 @@ public void testFirstErrorMsgOnReplay() { Assert.assertEquals("http://127.0.0.1/error_log", committedJob.getErrorLogUrls().peek()); Assert.assertEquals("invalid source row", committedJob.getFirstErrorMsg()); - long originalReserveHours = Config.load_error_log_reserve_hours; - try { - Config.load_error_log_reserve_hours = 0; - committedJob.processTimeoutTasks(); - Assert.assertTrue(committedJob.getErrorLogUrls().isEmpty()); - Assert.assertEquals("", committedJob.getFirstErrorMsg()); - } finally { - Config.load_error_log_reserve_hours = originalReserveHours; - } + RLTaskTxnCommitAttachment attachmentWithoutError = new RLTaskTxnCommitAttachment(); + Deencapsulation.setField(attachmentWithoutError, "progress", new KafkaProgress(thriftProgress)); + TransactionState cleanCommittedTxnState = Mockito.mock(TransactionState.class); + Mockito.doReturn(attachmentWithoutError).when(cleanCommittedTxnState).getTxnCommitAttachment(); + committedJob.replayOnCommitted(cleanCommittedTxnState); + Assert.assertTrue(committedJob.getErrorLogUrls().isEmpty()); + Assert.assertEquals("", committedJob.getFirstErrorMsg()); TransactionState abortedTxnState = Mockito.mock(TransactionState.class); Mockito.doReturn(attachment).when(abortedTxnState).getTxnCommitAttachment(); From e48bd7a9ebebce4af682271909e81e6e65435d99 Mon Sep 17 00:00:00 2001 From: laihui Date: Tue, 14 Jul 2026 11:11:35 +0800 Subject: [PATCH 07/11] [fix](load) Remove unrelated meta service diff --- cloud/src/meta-service/meta_service_txn.cpp | 2 ++ 1 file changed, 2 insertions(+) diff --git a/cloud/src/meta-service/meta_service_txn.cpp b/cloud/src/meta-service/meta_service_txn.cpp index c5a475a4f4aea6..5c35958b1f0b83 100644 --- a/cloud/src/meta-service/meta_service_txn.cpp +++ b/cloud/src/meta-service/meta_service_txn.cpp @@ -578,6 +578,7 @@ void put_routine_load_progress(MetaServiceCode& code, std::string& msg, new_progress_info.mutable_partition_to_offset()->insert(elem); } } + std::string new_statistic_val; RoutineLoadJobStatisticPB* new_statistic_info = new_progress_info.mutable_stat(); if (prev_progress_info.has_stat()) { @@ -886,6 +887,7 @@ void MetaServiceImpl::reset_rl_progress(::google::protobuf::RpcController* contr } } } + if (!new_progress_info.SerializeToString(&new_progress_val)) { code = MetaServiceCode::PROTOBUF_SERIALIZE_ERR; ss << "failed to serialize new progress val" From b42378035e4baca8c06079962689d3e69515194f Mon Sep 17 00:00:00 2001 From: laihui Date: Tue, 14 Jul 2026 11:15:37 +0800 Subject: [PATCH 08/11] [fix](load) Align routine load error cleanup with statistics reset --- .../load/routineload/RoutineLoadJob.java | 20 +++++++++---------- .../load/routineload/RoutineLoadJobTest.java | 10 ++++++++-- 2 files changed, 18 insertions(+), 12 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 01e7eeda8525db..f0d7dc36332280 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 @@ -960,10 +960,7 @@ private void updateNumOfData(long numOfTotalRows, long numOfErrorRows, long unse + "when current total rows is more than base or the filter ratio is more than the max") .build()); } - // reset currentTotalNum, currentErrorNum and otherMsg - this.jobStatistic.currentErrorRows = 0; - this.jobStatistic.currentTotalRows = 0; - this.otherMsg = ""; + resetCurrentErrorStatistics(); this.jobStatistic.currentAbortedTaskNum = 0; } else if (this.jobStatistic.currentErrorRows > maxErrorNum || (this.jobStatistic.currentTotalRows > 0 @@ -983,13 +980,18 @@ private void updateNumOfData(long numOfTotalRows, long numOfErrorRows, long unse "current error rows is more than max_error_number " + "or the max_filter_ratio is more than the value set"), isReplay); } - // reset currentTotalNum, currentErrorNum and otherMsg - this.jobStatistic.currentErrorRows = 0; - this.jobStatistic.currentTotalRows = 0; - this.otherMsg = ""; + resetCurrentErrorStatistics(); } } + private void resetCurrentErrorStatistics() { + this.jobStatistic.currentErrorRows = 0; + this.jobStatistic.currentTotalRows = 0; + this.otherMsg = ""; + this.errorLogUrls.clear(); + this.firstErrorMsg = ""; + } + protected void replayUpdateProgress(RLTaskTxnCommitAttachment attachment) { try { updateNumOfData(attachment.getTotalRows(), attachment.getFilteredRows(), attachment.getUnselectedRows(), @@ -1438,8 +1440,6 @@ private void executeTaskOnTxnStatusChanged(RoutineLoadTaskInfo routineLoadTaskIn private void updateJobInfoFromAttachment(RLTaskTxnCommitAttachment attachment) { if (Strings.isNullOrEmpty(attachment.getErrorLogUrl())) { - errorLogUrls.clear(); - firstErrorMsg = ""; return; } errorLogUrls.add(attachment.getErrorLogUrl()); diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java index fea839ae018bc5..9ef5aeb164ffbb 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java @@ -102,8 +102,8 @@ public void testFirstErrorMsgOnReplay() { TransactionState cleanCommittedTxnState = Mockito.mock(TransactionState.class); Mockito.doReturn(attachmentWithoutError).when(cleanCommittedTxnState).getTxnCommitAttachment(); committedJob.replayOnCommitted(cleanCommittedTxnState); - Assert.assertTrue(committedJob.getErrorLogUrls().isEmpty()); - Assert.assertEquals("", committedJob.getFirstErrorMsg()); + Assert.assertEquals("http://127.0.0.1/error_log", committedJob.getErrorLogUrls().peek()); + Assert.assertEquals("invalid source row", committedJob.getFirstErrorMsg()); TransactionState abortedTxnState = Mockito.mock(TransactionState.class); Mockito.doReturn(attachment).when(abortedTxnState).getTxnCommitAttachment(); @@ -345,11 +345,17 @@ public void testUpdateTotalMoreThanBatch() { RoutineLoadStatistic jobStatistic = Deencapsulation.getField(routineLoadJob, "jobStatistic"); Deencapsulation.setField(jobStatistic, "currentErrorRows", 1); Deencapsulation.setField(jobStatistic, "currentTotalRows", 99); + Deencapsulation.setField(routineLoadJob, "otherMsg", "previous warning"); + routineLoadJob.getErrorLogUrls().add("http://127.0.0.1/error_log"); + Deencapsulation.setField(routineLoadJob, "firstErrorMsg", "invalid source row"); Deencapsulation.invoke(routineLoadJob, "updateNumOfData", 2L, 0L, 0L, 1L, 1L, false); Assert.assertEquals(RoutineLoadJob.JobState.RUNNING, Deencapsulation.getField(routineLoadJob, "state")); Assert.assertEquals(new Long(0), Deencapsulation.getField(jobStatistic, "currentErrorRows")); Assert.assertEquals(new Long(0), Deencapsulation.getField(jobStatistic, "currentTotalRows")); + Assert.assertEquals("", Deencapsulation.getField(routineLoadJob, "otherMsg")); + Assert.assertTrue(routineLoadJob.getErrorLogUrls().isEmpty()); + Assert.assertEquals("", routineLoadJob.getFirstErrorMsg()); } From 593b0fc73f44bd9c9fecda482cda862c9518cfee Mon Sep 17 00:00:00 2001 From: laihui Date: Tue, 14 Jul 2026 11:21:03 +0800 Subject: [PATCH 09/11] [fix](load) Keep routine load errors out of replay --- .../load/routineload/RoutineLoadJob.java | 16 ++++----- .../load/routineload/RoutineLoadJobTest.java | 36 ------------------- 2 files changed, 6 insertions(+), 46 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 f0d7dc36332280..83dd90ea1c9599 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 @@ -1206,9 +1206,7 @@ public void afterCommitted(TransactionState txnState, boolean txnOperated) throw @Override public void replayOnCommitted(TransactionState txnState) { Preconditions.checkNotNull(txnState.getTxnCommitAttachment(), txnState); - RLTaskTxnCommitAttachment attachment = (RLTaskTxnCommitAttachment) txnState.getTxnCommitAttachment(); - replayUpdateProgress(attachment); - updateJobInfoFromAttachment(attachment); + replayUpdateProgress((RLTaskTxnCommitAttachment) txnState.getTxnCommitAttachment()); this.jobStatistic.committedTaskNum++; if (LOG.isDebugEnabled()) { LOG.debug("replay on committed: {}", txnState); @@ -1383,13 +1381,11 @@ public void replayOnAborted(TransactionState txnState) { // it need check commit info before update progress // for follower FE node progress may exceed correct progress // the data will lost if FE leader change at this moment - if (txnState.getTxnCommitAttachment() != null) { - RLTaskTxnCommitAttachment attachment = (RLTaskTxnCommitAttachment) txnState.getTxnCommitAttachment(); - if (checkCommitInfo(attachment, txnState, - TransactionState.TxnStatusChangeReason.fromString(txnState.getReason()))) { - replayUpdateProgress(attachment); - } - updateJobInfoFromAttachment(attachment); + if (txnState.getTxnCommitAttachment() != null + && checkCommitInfo((RLTaskTxnCommitAttachment) txnState.getTxnCommitAttachment(), + txnState, + TransactionState.TxnStatusChangeReason.fromString(txnState.getReason()))) { + replayUpdateProgress((RLTaskTxnCommitAttachment) txnState.getTxnCommitAttachment()); } this.jobStatistic.abortedTaskNum++; if (LOG.isDebugEnabled()) { diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java index 9ef5aeb164ffbb..263515975437fa 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java @@ -42,7 +42,6 @@ import org.apache.doris.transaction.GlobalTransactionMgrIface; import org.apache.doris.transaction.TransactionException; import org.apache.doris.transaction.TransactionState; -import org.apache.doris.transaction.TransactionStatus; import org.apache.doris.transaction.TxnStateCallbackFactory; import com.google.common.base.Strings; @@ -80,41 +79,6 @@ public void testFirstErrorMsgInTxnCommitAttachment() { Assert.assertEquals(attachment.getFirstErrorMsg(), cloudAttachment.getFirstErrorMsg()); } - @Test - public void testFirstErrorMsgOnReplay() { - TKafkaRLTaskProgress thriftProgress = new TKafkaRLTaskProgress(Maps.newHashMap()); - RLTaskTxnCommitAttachment attachment = new RLTaskTxnCommitAttachment(); - Deencapsulation.setField(attachment, "progress", new KafkaProgress(thriftProgress)); - Deencapsulation.setField(attachment, "errorLogUrl", "http://127.0.0.1/error_log"); - Deencapsulation.setField(attachment, "firstErrorMsg", "invalid source row"); - - TransactionState committedTxnState = Mockito.mock(TransactionState.class); - Mockito.doReturn(attachment).when(committedTxnState).getTxnCommitAttachment(); - Mockito.when(committedTxnState.getTransactionStatus()).thenReturn(TransactionStatus.COMMITTED); - RoutineLoadJob committedJob = new KafkaRoutineLoadJob(); - Deencapsulation.setField(committedJob, "progress", new KafkaProgress(thriftProgress)); - committedJob.replayOnCommitted(committedTxnState); - Assert.assertEquals("http://127.0.0.1/error_log", committedJob.getErrorLogUrls().peek()); - Assert.assertEquals("invalid source row", committedJob.getFirstErrorMsg()); - - RLTaskTxnCommitAttachment attachmentWithoutError = new RLTaskTxnCommitAttachment(); - Deencapsulation.setField(attachmentWithoutError, "progress", new KafkaProgress(thriftProgress)); - TransactionState cleanCommittedTxnState = Mockito.mock(TransactionState.class); - Mockito.doReturn(attachmentWithoutError).when(cleanCommittedTxnState).getTxnCommitAttachment(); - committedJob.replayOnCommitted(cleanCommittedTxnState); - Assert.assertEquals("http://127.0.0.1/error_log", committedJob.getErrorLogUrls().peek()); - Assert.assertEquals("invalid source row", committedJob.getFirstErrorMsg()); - - TransactionState abortedTxnState = Mockito.mock(TransactionState.class); - Mockito.doReturn(attachment).when(abortedTxnState).getTxnCommitAttachment(); - Mockito.when(abortedTxnState.getTransactionStatus()).thenReturn(TransactionStatus.ABORTED); - Mockito.when(abortedTxnState.getReason()).thenReturn("test abort"); - RoutineLoadJob abortedJob = new KafkaRoutineLoadJob(); - abortedJob.replayOnAborted(abortedTxnState); - Assert.assertEquals("http://127.0.0.1/error_log", abortedJob.getErrorLogUrls().peek()); - Assert.assertEquals("invalid source row", abortedJob.getFirstErrorMsg()); - } - @Test public void testAfterAbortedReasonOffsetOutOfRange() throws UserException { Env env = Mockito.mock(Env.class); From 7e56c5f51f6fca0f16ab739c03c3d43cde398866 Mon Sep 17 00:00:00 2001 From: laihui Date: Tue, 14 Jul 2026 11:24:16 +0800 Subject: [PATCH 10/11] [fix](load) Inline routine load error update --- .../doris/load/routineload/RoutineLoadJob.java | 17 ++++++----------- 1 file changed, 6 insertions(+), 11 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 83dd90ea1c9599..78b488bf756028 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 @@ -1416,8 +1416,12 @@ private void executeTaskOnTxnStatusChanged(RoutineLoadTaskInfo routineLoadTaskIn routineLoadTaskInfo.handleTaskByTxnCommitAttachment(rlTaskTxnCommitAttachment); } - if (rlTaskTxnCommitAttachment != null) { - updateJobInfoFromAttachment(rlTaskTxnCommitAttachment); + if (rlTaskTxnCommitAttachment != null + && !Strings.isNullOrEmpty(rlTaskTxnCommitAttachment.getErrorLogUrl())) { + errorLogUrls.add(rlTaskTxnCommitAttachment.getErrorLogUrl()); + firstErrorMsg = StringUtils.abbreviate( + Strings.nullToEmpty(rlTaskTxnCommitAttachment.getFirstErrorMsg()), + Config.first_error_msg_max_length); } routineLoadTaskInfo.setTxnStatus(txnStatus); @@ -1434,15 +1438,6 @@ private void executeTaskOnTxnStatusChanged(RoutineLoadTaskInfo routineLoadTaskIn } } - private void updateJobInfoFromAttachment(RLTaskTxnCommitAttachment attachment) { - if (Strings.isNullOrEmpty(attachment.getErrorLogUrl())) { - return; - } - errorLogUrls.add(attachment.getErrorLogUrl()); - firstErrorMsg = StringUtils.abbreviate( - Strings.nullToEmpty(attachment.getFirstErrorMsg()), Config.first_error_msg_max_length); - } - protected static void checkMeta(OlapTable olapTable, RoutineLoadDesc routineLoadDesc) throws UserException { if (routineLoadDesc == null) { return; From a08f5a2169ac1b9a2ae67e31fbcfff0eb5d04f7f Mon Sep 17 00:00:00 2001 From: laihui Date: Tue, 14 Jul 2026 11:28:48 +0800 Subject: [PATCH 11/11] [fix](load) Avoid duplicate first error truncation --- .../org/apache/doris/load/routineload/RoutineLoadJob.java | 4 +--- 1 file changed, 1 insertion(+), 3 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 78b488bf756028..9873368f405114 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 @@ -1419,9 +1419,7 @@ private void executeTaskOnTxnStatusChanged(RoutineLoadTaskInfo routineLoadTaskIn if (rlTaskTxnCommitAttachment != null && !Strings.isNullOrEmpty(rlTaskTxnCommitAttachment.getErrorLogUrl())) { errorLogUrls.add(rlTaskTxnCommitAttachment.getErrorLogUrl()); - firstErrorMsg = StringUtils.abbreviate( - Strings.nullToEmpty(rlTaskTxnCommitAttachment.getFirstErrorMsg()), - Config.first_error_msg_max_length); + firstErrorMsg = Strings.nullToEmpty(rlTaskTxnCommitAttachment.getFirstErrorMsg()); } routineLoadTaskInfo.setTxnStatus(txnStatus);