[feat](job) show first error message for routine load

be/src/information_schema/schema_routine_load_job_scanner.cpp

@@ -55,6 +55,7 @@ std::vector<SchemaScanner::ColumnDesc> SchemaRoutineLoadJobScanner::_s_tbls_colu
         {"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()
@@ -175,6 +176,9 @@ Status SchemaRoutineLoadJobScanner::_fill_block_impl(Block* block) {
                 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] =

be/src/load/stream_load/stream_load_executor.cpp

@@ -110,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);
@@ -118,7 +121,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 +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);
+        if (!ctx->first_error_msg.empty()) {
+            rl_attach.__set_firstErrorMsg(ctx->first_error_msg);
+        }
 
         attach->rlTaskTxnCommitAttachment = rl_attach;
         attach->__isset.rlTaskTxnCommitAttachment = true;

fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java

@@ -706,6 +706,7 @@ public class SchemaTable extends Table {
                                     .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",

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;

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();

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": ""}
@@ -46,6 +48,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 +78,9 @@ public RLTaskTxnCommitAttachment(TRLTaskTxnCommitAttachment rlTaskTxnCommitAttac
         if (rlTaskTxnCommitAttachment.isSetErrorLogUrl()) {
             this.errorLogUrl = rlTaskTxnCommitAttachment.getErrorLogUrl();
         }
+        if (rlTaskTxnCommitAttachment.isSetFirstErrorMsg()) {
+            this.firstErrorMsg = abbreviateFirstErrorMsg(rlTaskTxnCommitAttachment.getFirstErrorMsg());
+        }
     }
 
     public RLTaskTxnCommitAttachment(RLTaskTxnCommitAttachmentPB rlTaskTxnCommitAttachment) {
@@ -92,6 +98,11 @@ public RLTaskTxnCommitAttachment(RLTaskTxnCommitAttachmentPB rlTaskTxnCommitAtta
 
         this.progress = progress;
         this.errorLogUrl = rlTaskTxnCommitAttachment.getErrorLogUrl();
+        this.firstErrorMsg = abbreviateFirstErrorMsg(rlTaskTxnCommitAttachment.getFirstErrorMsg());
+    }
+
+    private static String abbreviateFirstErrorMsg(String firstErrorMsg) {
+        return StringUtils.abbreviate(firstErrorMsg, Config.first_error_msg_max_length);
     }
 
     public long getJobId() {
@@ -134,6 +145,10 @@ public String getErrorLogUrl() {
         return errorLogUrl;
     }
 
+    public String getFirstErrorMsg() {
+        return firstErrorMsg;
+    }
+
     @Override
     public String toString() {
         return "RLTaskTxnCommitAttachment [filteredRows=" + filteredRows

fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java

@@ -274,8 +274,10 @@ public boolean isFinalState() {
     protected Expr deleteCondition;
     // TODO(ml): error sample
 
-    // save the latest 3 error log urls
-    private Queue<String> errorLogUrls = EvictingQueue.create(3);
+    // 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<String> errorLogUrls = EvictingQueue.create(3);
+    private transient String firstErrorMsg = "";
 
     @SerializedName("ccid")
     private String cloudClusterId;
@@ -809,6 +811,10 @@ public Queue<String> 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();
@@ -954,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
@@ -977,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(),
@@ -1408,8 +1416,10 @@ private void executeTaskOnTxnStatusChanged(RoutineLoadTaskInfo routineLoadTaskIn
             routineLoadTaskInfo.handleTaskByTxnCommitAttachment(rlTaskTxnCommitAttachment);
         }
 
-        if (rlTaskTxnCommitAttachment != null && !Strings.isNullOrEmpty(rlTaskTxnCommitAttachment.getErrorLogUrl())) {
+        if (rlTaskTxnCommitAttachment != null
+                && !Strings.isNullOrEmpty(rlTaskTxnCommitAttachment.getErrorLogUrl())) {
             errorLogUrls.add(rlTaskTxnCommitAttachment.getErrorLogUrl());
+            firstErrorMsg = Strings.nullToEmpty(rlTaskTxnCommitAttachment.getFirstErrorMsg());
         }
 
         routineLoadTaskInfo.setTxnStatus(txnStatus);
@@ -1712,6 +1722,7 @@ public List<String> getShowInfo() {
             row.add(userIdentity.getQualifiedUser());
             row.add(comment);
             row.add(getClusterInfo());
+            row.add(getFirstErrorMsg());
             return row;
         } finally {
             readUnlock();
@@ -1951,6 +1962,8 @@ public void write(DataOutput out) throws IOException {
 
     @Override
     public void gsonPostProcess() throws IOException {
+        errorLogUrls = EvictingQueue.create(3);
+        firstErrorMsg = "";
         if (tableId == 0) {
             isMultiTable = true;
         }

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;

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());

fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgrTest.java

@@ -36,15 +36,18 @@
 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;
 import org.junit.Assert;
 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<MetaServiceProxy> 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<TransactionState> 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);