Skip to content
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

Jump to bottom

[branch-2.1][Feat](Job)After a job is paused, it can be manually triggered to execute. #40180

Merged
merged 1 commit into from
Aug 31, 2024
Merged

[branch-2.1][Feat](Job)After a job is paused, it can be manually triggered to execute. #40180

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
26 changes: 20 additions & 6 deletions fe/fe-core/src/main/java/org/apache/doris/job/base/AbstractJob.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -212,8 +212,9 @@ public List<T> queryAllTasks() {
}

public List<T> commonCreateTasks(TaskType taskType, C taskContext) {
if (!getJobStatus().equals(JobStatus.RUNNING)) {
log.warn("job is not running, job id is {}", jobId);
if (!canCreateTask(taskType)) {
log.info("job is not ready for scheduling, job id is {},job status is {}, taskType is {}", jobId,
jobStatus, taskType);
return new ArrayList<>();
}
if (!isReadyForScheduling(taskContext)) {
Expand All @@ -235,6 +236,19 @@ public List<T> commonCreateTasks(TaskType taskType, C taskContext) {
}
}

private boolean canCreateTask(TaskType taskType) {
JobStatus currentJobStatus = getJobStatus();

switch (taskType) {
case SCHEDULED:
return currentJobStatus.equals(JobStatus.RUNNING);
case MANUAL:
return currentJobStatus.equals(JobStatus.RUNNING) || currentJobStatus.equals(JobStatus.PAUSED);
default:
throw new IllegalArgumentException("Unsupported TaskType: " + taskType);
}
}

public void initTasks(Collection<? extends T> tasks, TaskType taskType) {
tasks.forEach(task -> {
task.setTaskType(taskType);
Expand Down Expand Up @@ -307,24 +321,24 @@ public void logUpdateOperation() {
@Override
public void onTaskFail(T task) throws JobException {
failedTaskCount.incrementAndGet();
updateJobStatusIfEnd(false);
updateJobStatusIfEnd(false, task.getTaskType());
runningTasks.remove(task);
logUpdateOperation();
}

@Override
public void onTaskSuccess(T task) throws JobException {
succeedTaskCount.incrementAndGet();
updateJobStatusIfEnd(true);
updateJobStatusIfEnd(true, task.getTaskType());
runningTasks.remove(task);
logUpdateOperation();

}


private void updateJobStatusIfEnd(boolean taskSuccess) throws JobException {
private void updateJobStatusIfEnd(boolean taskSuccess, TaskType taskType) throws JobException {
JobExecuteType executeType = getJobConfig().getExecuteType();
if (executeType.equals(JobExecuteType.MANUAL)) {
if (executeType.equals(JobExecuteType.MANUAL) || taskType.equals(TaskType.MANUAL)) {
return;
}
switch (executeType) {
Expand Down
8 changes: 7 additions & 1 deletion fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/InsertTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@ public class InsertTask extends AbstractTask {
new Column("Status", ScalarType.createStringType()),
new Column("ErrorMsg", ScalarType.createStringType()),
new Column("CreateTime", ScalarType.createStringType()),
new Column("StartTime", ScalarType.createStringType()),
new Column("FinishTime", ScalarType.createStringType()),
new Column("TrackingUrl", ScalarType.createStringType()),
new Column("LoadStatistic", ScalarType.createStringType()),
Expand Down Expand Up @@ -247,6 +248,8 @@ public TRow getTvfInfo(String jobName) {
trow.addToColumnValue(new TCell().setStringVal(errorMsg));
// create time
trow.addToColumnValue(new TCell().setStringVal(TimeUtils.longToTimeString(getCreateTimeMs())));
trow.addToColumnValue(new TCell().setStringVal(null == getStartTimeMs() ? ""
: TimeUtils.longToTimeString(getStartTimeMs())));
// load end time
trow.addToColumnValue(new TCell().setStringVal(TimeUtils.longToTimeString(getFinishTimeMs())));
// tracking url
Expand Down Expand Up @@ -274,7 +277,10 @@ private TRow getPendingTaskTVFInfo(String jobName) {
trow.addToColumnValue(new TCell().setStringVal(getStatus().name()));
trow.addToColumnValue(new TCell().setStringVal(""));
trow.addToColumnValue(new TCell().setStringVal(TimeUtils.longToTimeString(getCreateTimeMs())));
trow.addToColumnValue(new TCell().setStringVal(""));
trow.addToColumnValue(new TCell().setStringVal(null == getStartTimeMs() ? ""
: TimeUtils.longToTimeString(getStartTimeMs())));
trow.addToColumnValue(new TCell().setStringVal(null == getFinishTimeMs() ? ""
: TimeUtils.longToTimeString(getFinishTimeMs())));
trow.addToColumnValue(new TCell().setStringVal(""));
trow.addToColumnValue(new TCell().setStringVal(""));
trow.addToColumnValue(new TCell().setStringVal(userIdentity.getQualifiedUser()));
Expand Down
5 changes: 2 additions & 3 deletions fe/fe-core/src/main/java/org/apache/doris/job/scheduler/JobScheduler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,10 +188,9 @@ private void executeTimerJobIdsWithinLastTenMinutesWindow() {
clearEndJob(job);
continue;
}
if (!job.getJobStatus().equals(JobStatus.RUNNING) && !job.getJobConfig().checkIsTimerJob()) {
continue;
if (job.getJobStatus().equals(JobStatus.RUNNING) && job.getJobConfig().checkIsTimerJob()) {
cycleTimerJobScheduler(job, lastTimeWindowMs);
}
cycleTimerJobScheduler(job, lastTimeWindowMs);
}
}

Expand Down
Loading