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..0e954de76466bf 100644 --- a/be/src/information_schema/schema_routine_load_job_scanner.cpp +++ b/be/src/information_schema/schema_routine_load_job_scanner.cpp @@ -55,6 +55,7 @@ std::vector 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] = diff --git a/be/src/load/stream_load/stream_load_executor.cpp b/be/src/load/stream_load/stream_load_executor.cpp index 478ed22755df9e..9b307c932d4776 100644 --- a/be/src/load/stream_load/stream_load_executor.cpp +++ b/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; 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..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 @@ -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", 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/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..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": ""} @@ -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 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..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 @@ -274,8 +274,10 @@ public boolean isFinalState() { protected Expr deleteCondition; // TODO(ml): error sample - // save the latest 3 error log urls - private Queue 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 errorLogUrls = EvictingQueue.create(3); + private transient 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(); @@ -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 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; } 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/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); 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..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 @@ -21,6 +21,8 @@ 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.Config; import org.apache.doris.common.InternalErrorCode; import org.apache.doris.common.Pair; import org.apache.doris.common.UserException; @@ -33,6 +35,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 +58,27 @@ import java.util.Map; 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(overlongFirstErrorMsg); + + RLTaskTxnCommitAttachment attachment = new RLTaskTxnCommitAttachment(thriftAttachment); + 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(attachment.getFirstErrorMsg(), cloudAttachment.getFirstErrorMsg()); + } + @Test public void testAfterAbortedReasonOffsetOutOfRange() throws UserException { Env env = Mockito.mock(Env.class); @@ -95,6 +121,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 +142,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 +183,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 @@ -276,11 +309,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()); } 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..e5c458132a94e1 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; + // Error log URLs and first error messages are transient diagnostics and are not stored here. } message RLTaskTxnCommitAttachmentPB { @@ -406,6 +407,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..a41c84d2ce04a0 --- /dev/null +++ b/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 """ + 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 test_routine_load_first_error_msg" + 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("bad_row") + && showFirstErrorMsg != null + && showFirstErrorMsg.toString().contains("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}" + } +}