github-actions[bot] commented on code in PR #65553:
URL: https://github.com/apache/doris/pull/65553#discussion_r3572005072
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java:
##########
@@ -1408,6 +1414,12 @@ private void
executeTaskOnTxnStatusChanged(RoutineLoadTaskInfo routineLoadTaskIn
routineLoadTaskInfo.handleTaskByTxnCommitAttachment(rlTaskTxnCommitAttachment);
}
Review Comment:
This only updates `firstErrorMsg` in the live transaction callback. On FE
restart or leader failover, replay goes through `replayOnCommitted()` /
`replayOnAborted()` and the replay path only restores offsets/statistics; it
never copies `attachment.getFirstErrorMsg()` into the job. That also matters
for aborted task attachments: the live path records this display field even
when `checkCommitInfo()` rejects a progress update, so putting the copy only
inside `replayUpdateProgress()` would still miss that case. Please apply the
attachment display fields during replay wherever the live path would record
them, while keeping progress/offset updates gated by the existing checks, and
add replay coverage.
##########
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())
Review Comment:
Adding this column here shifts the existing `USER_NAME`,
`CURRENT_ABORT_TASK_NUM`, `IS_ABNORMAL_PAUSE`, and `COMPUTE_GROUP` ordinals in
`information_schema.routine_load_jobs`. The other public surfaces in this PR
append the new field, so this is an avoidable compatibility break for `SELECT
*` or ordinal-based clients. Please append `FIRST_ERROR_MSG` after
`COMPUTE_GROUP` in both `SchemaTable` and the BE scanner instead.
##########
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());
Review Comment:
This sets `first_error_msg` on the cloud commit attachment, but the cloud
progress path still drops it. `put_routine_load_progress()` stores only
`RoutineLoadProgressPB` progress/statistics, and `get_rl_task_commit_attach()`
reconstructs a response attachment from that blob without `first_error_msg`;
`KafkaRoutineLoadJob.updateCloudProgress()` then has no value to apply. In
cloud mode the new SHOW/information_schema field disappears after meta-service
refresh/restart. Please persist and return the message at the routine-load
progress boundary, and apply it during `updateCloudProgress()`.
##########
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"
Review Comment:
This is an ordinary single-table regression case, so it should follow the
local test standard by hardcoding `test_routine_load_first_error_msg` in the
SQL instead of routing it through `def tableName`. The same standard also asks
tests to drop tables before use rather than after completion; this file already
does the pre-test drop, so the final `DROP TABLE` should be removed to preserve
failed-test state for debugging.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
