github-actions[bot] commented on code in PR #66105:
URL: https://github.com/apache/doris/pull/66105#discussion_r3656242545
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java:
##########
@@ -244,41 +244,28 @@ private void scheduleOneTask(RoutineLoadTaskInfo
routineLoadTaskInfo) throws Exc
routineLoadTaskInfo.setExecuteStartTimeMs(System.currentTimeMillis());
}
- private void handleSubmitTaskFailure(RoutineLoadTaskInfo
routineLoadTaskInfo, String errorMsg) {
+ private void handleSubmitTaskFailure(RoutineLoadTaskInfo
routineLoadTaskInfo, String errorMsg)
+ throws UserException {
LOG.warn("failed to submit routine load task {} to BE: {}, error: {}",
DebugUtil.printId(routineLoadTaskInfo.getId()),
routineLoadTaskInfo.getBeId(), errorMsg);
routineLoadTaskInfo.setBeId(-1);
RoutineLoadJob routineLoadJob =
routineLoadManager.getJob(routineLoadTaskInfo.getJobId());
- RoutineLoadTaskInfo newTask;
routineLoadJob.writeLock();
try {
- routineLoadJob.setOtherMsg(errorMsg);
-
- // Check if this is a resource pressure error that should not be
immediately rescheduled
- if (errorMsg.contains("TOO_MANY_TASKS") ||
errorMsg.contains("MEM_LIMIT_EXCEEDED")) {
- // submit task failed (such as
TOO_MANY_TASKS/MEM_LIMIT_EXCEEDED error),
- // but txn has already begun. Here we will still set the
ExecuteStartTime of
- // this task, which means we "assume" that this task has been
successfully submitted.
- // And this task will then be aborted because of a timeout.
- // In this way, we can prevent the entire job from being
paused due to submit errors,
- // and we can also relieve the pressure on BE by waiting for
the timeout period.
-
routineLoadTaskInfo.setExecuteStartTimeMs(System.currentTimeMillis());
- return;
- }
-
if (routineLoadJob.getState() != JobState.RUNNING
||
!routineLoadJob.containsTask(routineLoadTaskInfo.getId())) {
return;
}
- // for other errors (network issues, BE restart, etc.), reschedule
immediately
- newTask = routineLoadJob.unprotectRenewTask(routineLoadTaskInfo,
false);
+ routineLoadJob.unprotectUpdateState(JobState.PAUSED,
Review Comment:
[P1] Close the begun transaction before auto-resuming
`beginTxn()` has already created a PREPARE transaction before this path
runs, but `executePause()` only clears the task list. For real
`TOO_MANY_TASKS`/`MEM_LIMIT_EXCEEDED` responses, BE rejects before registering
the task, so no task callback will terminate that FE transaction.
`CREATE_TASKS_ERR` can then auto-resume and create replacements before timeout
cleanup; with the default non-EOF Kafka timeout, one continuously failing job
can leave roughly 15 rejected PREPARE transactions live, plus any peer
transactions cleared by the job-wide pause.
These transactions count against the database quota, so enough failing jobs
(or a lower supported quota) can reject unrelated writes; after that,
`beginTxn()` requeues the task at the head of the scheduler's 1 ms loop. Please
terminate the captured transaction outside the job lock, or gate replacement
scheduling until it is terminal, and add a persistent-failure test that
verifies the transaction count stays bounded.
--
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]
