github-actions[bot] commented on code in PR #65553:
URL: https://github.com/apache/doris/pull/65553#discussion_r3575825419
##########
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);
Review Comment:
This fixes the OK cloud abort path, but the idempotent retry path still
loses the rollback attachment. If meta-service already processed the abort and
a retry returns `TXN_ALREADY_ABORTED`, `abortTransactionImpl()` calls
`afterAbortTxnResp()` before returning, so this assignment is skipped and the
`finally` calls `handleAfterAbort(null, ...)`. That callback sees an empty,
non-operated `TransactionState`, so `RoutineLoadJob.afterAborted()` never
records the `firstErrorMsg`, even though meta-service returns the aborted
`txn_info` with the request attachment copied into it. Please treat
`TXN_ALREADY_ABORTED` with returned txn info as an idempotent abort for the
callback, or otherwise pass the returned/requested attachment into
`handleAfterAbort()`, and add a cloud abort retry test.
##########
regression-test/suites/load_p0/routine_load/test_routine_load_first_error_msg.groovy:
##########
@@ -0,0 +1,127 @@
+// 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 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 test_routine_load_first_error_msg"
+ sql """
+ CREATE TABLE test_routine_load_first_error_msg (
+ 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 test_routine_load_first_error_msg
+ 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 """
Review Comment:
This query can pick up a stopped job from a previous run. The suite uses the
fixed routine-load name and only stops the job in `finally`; stopped jobs
remain until `label_keep_max_second` expires, while `isNameUsed()` ignores
final-state jobs so a rerun can create another job with the same name.
`information_schema.routine_load_jobs` is populated from
`getAllRoutineLoadJobs()` without filtering to the active job, and the test
reads `jobInfo[0]`, so it can compare the old row's URL/message against the new
`SHOW ROUTINE LOAD` result or time out. Please use a unique job name or filter
by the current job id/state before reading the information_schema row.
##########
be/src/load/stream_load/stream_load_executor.cpp:
##########
@@ -428,6 +430,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);
Review Comment:
`RuntimeState::append_error_msg_to_file()` records the first message as the
full error plus source line before the error-log-file byte cap, and this line
sends that raw string inside `TLoadTxnCommitRequest` /
`TLoadTxnRollbackRequest`. The new FE-side abbreviation in
`RLTaskTxnCommitAttachment` only runs after the Thrift request has already been
received, so one large malformed Kafka row can still make the routine-load
transaction RPC carry a very large payload or exceed FE thrift frame/message
limits before the durable/cloud cap protects anything. Please cap
`ctx->first_error_msg` before setting
`TRLTaskTxnCommitAttachment.firstErrorMsg` or otherwise at the BE RPC
serialization boundary, and cover an overlong routine-load bad row.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java:
##########
@@ -1712,6 +1724,7 @@ public List<String> getShowInfo() {
row.add(userIdentity.getQualifiedUser());
row.add(comment);
row.add(getClusterInfo());
+ row.add(getFirstErrorMsg());
return row;
} finally {
readUnlock();
Review Comment:
This timestamps the diagnostic when FE processes the attachment, but the URL
lifetime is controlled by BE error-log file age. A routine-load job can legally
set `max_batch_interval` above the default 48-hour BE
`load_error_log_reserve_hours`; if the first bad row appears early and the
transaction callback arrives after BE cleanup, FE records a fresh timestamp
here and keeps showing a dead URL/message for another retention window. Replay
has the same problem because old attachments are repopulated with the replay
time. Please timestamp this from a value tied to the BE log lifetime, such as
task/error-log creation time or transaction finish time, and skip or expire
entries whose backing log may already be gone.
--
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]
