database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[feat](job)Internal job cancellation immediately and the strong association with the STARTS parameter (#36805) (#38110)

Browse Source 
…

## Proposed changes

For internal tasks, such as MTMV, the start time may already be set, or
the time may be adjusted immediately.

<!--Describe your changes.-->

(cherry picked from commit 904a6c0fc1a804520285533de874fe4d0ffff2c1)

## Proposed changes

Issue Number: close #36805

<!--Describe your changes.-->
This commit is contained in:
Calvin Kirs
2024-09-05 16:28:35 +08:00
committed by GitHub
parent 52393f829d
commit 961d2c9af5
5 changed files with 12 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
fe/fe-core/src/main/java/org/apache/doris/analysis/CreateJobStmt.java
View File

@ -128,6 +128,7 @@ public class CreateJobStmt extends DdlStmt {
if (null != onceJobStartTimestamp) {
if (onceJobStartTimestamp.equalsIgnoreCase(CURRENT_TIMESTAMP_STRING)) {
jobExecutionConfiguration.setImmediate(true);
timerDefinition.setStartTimeMs(System.currentTimeMillis());
} else {
timerDefinition.setStartTimeMs(TimeUtils.timeStringToLong(onceJobStartTimestamp));
}
@ -149,6 +150,8 @@ public class CreateJobStmt extends DdlStmt {
if (null != startsTimeStamp) {
if (startsTimeStamp.equalsIgnoreCase(CURRENT_TIMESTAMP_STRING)) {
jobExecutionConfiguration.setImmediate(true);
//To avoid immediate re-scheduling, set the start time of the timer 100ms before the current time.
timerDefinition.setStartTimeMs(System.currentTimeMillis());
} else {
timerDefinition.setStartTimeMs(TimeUtils.timeStringToLong(startsTimeStamp));
}

8
fe/fe-core/src/main/java/org/apache/doris/job/base/JobExecutionConfiguration.java
View File

@ -57,9 +57,7 @@ public class JobExecutionConfiguration {
if (executeType == JobExecuteType.INSTANT || executeType == JobExecuteType.MANUAL) {
return;
}
checkTimerDefinition(immediate);
checkTimerDefinition();
if (executeType == JobExecuteType.ONE_TIME) {
validateStartTimeMs();
return;
@ -80,12 +78,12 @@ public class JobExecutionConfiguration {
}
}
private void checkTimerDefinition(boolean immediate) {
private void checkTimerDefinition() {
if (timerDefinition == null) {
throw new IllegalArgumentException(
"timerDefinition cannot be null when executeType is not instant or manual");
}
timerDefinition.checkParams(immediate);
timerDefinition.checkParams();
}
private void validateStartTimeMs() {

8
fe/fe-core/src/main/java/org/apache/doris/job/base/TimerDefinition.java
View File

@ -38,13 +38,7 @@ public class TimerDefinition {
private Long latestSchedulerTimeMs;
public void checkParams(boolean immediate) {
if (null != startTimeMs && immediate) {
throw new IllegalArgumentException("startTimeMs must be null when immediate is true");
}
if (null == startTimeMs && immediate) {
startTimeMs = System.currentTimeMillis();
}
public void checkParams() {
if (null == startTimeMs) {
startTimeMs = System.currentTimeMillis() + intervalUnit.getIntervalMs(interval);
}

4
fe/fe-core/src/main/java/org/apache/doris/job/scheduler/JobScheduler.java
View File

@ -124,6 +124,10 @@ public class JobScheduler<T extends AbstractJob<?, C>, C> implements Closeable {
schedulerInstantJob(job, TaskType.SCHEDULED, null);
}
}
if (job.getJobConfig().isImmediate() && JobExecuteType.ONE_TIME.equals(job.getJobConfig().getExecuteType())) {
schedulerInstantJob(job, TaskType.SCHEDULED, null);
return;
}
//RECURRING job and immediate is true
if (job.getJobConfig().isImmediate()) {
job.getJobConfig().getTimerDefinition().setLatestSchedulerTimeMs(System.currentTimeMillis());