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

[Fix](Job)Fixed job scheduling missing certain time window schedules (#28659)

Browse Source 
Since scheduling itself consumes a certain amount of time, the start time of the time window should not be the current time, but the end time of the last schedule.
This commit is contained in:
Calvin Kirs
2023-12-20 09:21:15 +08:00
committed by GitHub
parent fb3b0afb85
commit 9aa878ea1f
3 changed files with 9 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -178,8 +178,8 @@ public class JobExecutionConfiguration {
// Calculate the trigger time list
for (long triggerTime = firstTriggerTime; triggerTime <= windowEndTimeMs; triggerTime += intervalMs) {
if (triggerTime >= currentTimeMs && (null == timerDefinition.getEndTimeMs()
|| triggerTime < timerDefinition.getEndTimeMs())) {
if (null == timerDefinition.getEndTimeMs()
|| triggerTime < timerDefinition.getEndTimeMs()) {
timerDefinition.setLatestSchedulerTimeMs(triggerTime);
timestamps.add(queryDelayTimeSecond(currentTimeMs, triggerTime));
}

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

@ -130,7 +130,7 @@ public class JobScheduler<T extends AbstractJob<?, C>, C> implements Closeable {
schedulerInstantJob(job, TaskType.SCHEDULED, null);
}
//if it's timer job and trigger last window already start, we will scheduler it immediately
cycleTimerJobScheduler(job);
cycleTimerJobScheduler(job, System.currentTimeMillis());
}
@Override
@ -139,9 +139,9 @@ public class JobScheduler<T extends AbstractJob<?, C>, C> implements Closeable {
}
private void cycleTimerJobScheduler(T job) {
private void cycleTimerJobScheduler(T job, long startTimeWindowMs) {
List<Long> delaySeconds = job.getJobConfig().getTriggerDelayTimes(System.currentTimeMillis(),
System.currentTimeMillis(), latestBatchSchedulerTimerTaskTimeMs);
startTimeWindowMs, latestBatchSchedulerTimerTaskTimeMs);
if (CollectionUtils.isNotEmpty(delaySeconds)) {
delaySeconds.forEach(delaySecond -> {
TimerJobSchedulerTask<T> timerJobSchedulerTask = new TimerJobSchedulerTask<>(timerJobDisruptor, job);
@ -170,6 +170,8 @@ public class JobScheduler<T extends AbstractJob<?, C>, C> implements Closeable {
* We will get the task in the next time window, and then hand it over to the time wheel for timing trigger
*/
private void executeTimerJobIdsWithinLastTenMinutesWindow() {
long lastTimeWindowMs = latestBatchSchedulerTimerTaskTimeMs;
if (latestBatchSchedulerTimerTaskTimeMs < System.currentTimeMillis()) {
this.latestBatchSchedulerTimerTaskTimeMs = System.currentTimeMillis();
}
@ -186,7 +188,7 @@ public class JobScheduler<T extends AbstractJob<?, C>, C> implements Closeable {
if (!job.getJobStatus().equals(JobStatus.RUNNING) && !job.getJobConfig().checkIsTimerJob()) {
continue;
}
cycleTimerJobScheduler(job);
cycleTimerJobScheduler(job, lastTimeWindowMs);
}
}