[feat](job)Internal job cancellation immediately and the strong assoc…

…iation with the STARTS parameter

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

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

@@ -128,6 +128,7 @@ public void analyze(Analyzer analyzer) throws UserException {
         if (null != onceJobStartTimestamp) {
             if (onceJobStartTimestamp.equalsIgnoreCase(CURRENT_TIMESTAMP_STRING)) {
                 jobExecutionConfiguration.setImmediate(true);
+                timerDefinition.setStartTimeMs(System.currentTimeMillis() - 100L);
             } else {
                 timerDefinition.setStartTimeMs(TimeUtils.timeStringToLong(onceJobStartTimestamp));
             }
@@ -149,6 +150,8 @@ public void analyze(Analyzer analyzer) throws UserException {
         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() - 100L);
             } else {
                 timerDefinition.setStartTimeMs(TimeUtils.timeStringToLong(startsTimeStamp));
             }

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

@@ -57,9 +57,7 @@ public void checkParams() {
         if (executeType == JobExecuteType.INSTANT || executeType == JobExecuteType.MANUAL) {
             return;
         }
-
-        checkTimerDefinition(immediate);
-
+        checkTimerDefinition();
         if (executeType == JobExecuteType.ONE_TIME) {
             validateStartTimeMs();
             return;
@@ -80,12 +78,12 @@ public void checkParams() {
         }
     }
 
-    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() {

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

@@ -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);
         }

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

@@ -124,6 +124,10 @@ public void scheduleOneJob(T job) throws JobException {
                 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());