Skip to content

branch-3.0: [fix](job) fix routine load task scheduler block for one job can not find any BE (#52654) #52791

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dataroaring merged 1 commit into from
Jul 8, 2025
Merged

branch-3.0: [fix](job) fix routine load task scheduler block for one job can not find any BE (#52654) #52791

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 11 additions & 1 deletion fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -498,8 +498,18 @@ public long getMinTaskBeId(String clusterName) throws LoadException {
// check if the specified BE is available for running task
// return true if it is available. return false if otherwise.
// throw exception if unrecoverable errors happen.
public long getAvailableBeForTask(long jobId, long previousBeId) throws LoadException {
public long getAvailableBeForTask(long jobId, long previousBeId) throws UserException {
List<Long> availableBeIds = getAvailableBackendIds(jobId);
if (availableBeIds.isEmpty()) {
RoutineLoadJob job = getJob(jobId);
if (job != null) {
String msg = "no available BE found for job " + jobId
+ "please check the BE status and user's cluster or tags";
job.updateState(RoutineLoadJob.JobState.PAUSED,
new ErrorReason(InternalErrorCode.MANUAL_PAUSE_ERR, msg), false /* not replay */);
}
return -1L;
}

// check if be has idle slot
readLock();
Expand Down
4 changes: 2 additions & 2 deletions fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -147,7 +147,7 @@ private void scheduleOneTask(RoutineLoadTaskInfo routineLoadTaskInfo) throws Exc
// this should be done before txn begin, or the txn may be begun successfully but failed to be allocated.
if (!allocateTaskToBe(routineLoadTaskInfo)) {
// allocate failed, push it back to the queue to wait next scheduling
needScheduleTasksQueue.addFirst(routineLoadTaskInfo);
needScheduleTasksQueue.addLast(routineLoadTaskInfo);
return;
}
} catch (UserException e) {
Expand Down Expand Up @@ -311,7 +311,7 @@ private void submitTask(long beId, TRoutineLoadTask tTask) throws LoadException
// 2. If not, try to find a better one with most idle slots.
// return true if allocate successfully. return false if failed.
// throw exception if unrecoverable errors happen.
private boolean allocateTaskToBe(RoutineLoadTaskInfo routineLoadTaskInfo) throws LoadException {
private boolean allocateTaskToBe(RoutineLoadTaskInfo routineLoadTaskInfo) throws UserException {
long beId = routineLoadManager.getAvailableBeForTask(routineLoadTaskInfo.getJobId(),
routineLoadTaskInfo.getPreviousBeId());
if (beId == -1L) {
Expand Down
Loading