ensure abnormal jobs do not interfere with normal task scheduling

Author: sollhui

fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java

@@ -371,18 +371,19 @@ protected void unprotectUpdateProgress() throws UserException {
     }
 
     @Override
-    protected void preCheckNeedSchedule() throws UserException {
+    protected boolean preCheckNeedSchedule(boolean isNeedAutoSchedule) throws UserException {
         // If user does not specify kafka partition,
         // We will fetch partition from kafka server periodically
-        if (this.state == JobState.RUNNING || this.state == JobState.NEED_SCHEDULE) {
+        if (this.state == JobState.RUNNING || this.state == JobState.NEED_SCHEDULE || isNeedAutoSchedule) {
             if (customKafkaPartitions != null && !customKafkaPartitions.isEmpty()) {
-                return;
+                return true;
             }
-            updateKafkaPartitions();
+            return updateKafkaPartitions();
         }
+        return true;
     }
 
-    private void updateKafkaPartitions() throws UserException {
+    private boolean updateKafkaPartitions() throws UserException {
         try {
             this.newCurrentKafkaPartition = getAllKafkaPartitions();
         } catch (Exception e) {
@@ -397,7 +398,9 @@ private void updateKafkaPartitions() throws UserException {
                         new ErrorReason(InternalErrorCode.PARTITIONS_ERR, msg),
                         false /* not replay */);
             }
+            return false;
         }
+        return true;
     }
 
     // if customKafkaPartition is not null, then return false immediately
@@ -406,7 +409,7 @@ private void updateKafkaPartitions() throws UserException {
     // update current kafka partition at the same time
     // current kafka partitions = customKafkaPartitions == 0 ? all partition of kafka topic : customKafkaPartitions
     @Override
-    protected boolean unprotectNeedReschedule() throws UserException {
+    protected boolean unprotectNeedReschedule(boolean isNeedAutoSchedule) throws UserException {
         // only running and need_schedule job need to be changed current kafka partitions
         if (this.state == JobState.RUNNING || this.state == JobState.NEED_SCHEDULE) {
             if (CollectionUtils.isNotEmpty(customKafkaPartitions)) {
@@ -447,11 +450,15 @@ protected boolean unprotectNeedReschedule() throws UserException {
             }
 
         }
+        return isNeedAutoSchedule;
+    }
+
+    @Override
+    protected boolean isNeedAutoSchedule() {
         if (this.state == JobState.PAUSED) {
             return ScheduleRule.isNeedAutoSchedule(this);
         }
         return false;
-
     }
 
     @Override

fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java

@@ -1510,11 +1510,15 @@ public void update() throws UserException {
             }
         }
 
-        preCheckNeedSchedule();
+        boolean isNeedAutoSchedule = isNeedAutoSchedule();
+
+        if (!preCheckNeedSchedule(isNeedAutoSchedule)) {
+            return;
+        }
 
         writeLock();
         try {
-            if (unprotectNeedReschedule()) {
+            if (unprotectNeedReschedule(isNeedAutoSchedule)) {
                 LOG.info(new LogBuilder(LogKey.ROUTINE_LOAD_JOB, id)
                         .add("msg", "Job need to be rescheduled")
                         .build());
@@ -1530,14 +1534,18 @@ public void update() throws UserException {
     // Because unprotectUpdateProgress() is protected by writelock.
     // So if there are time-consuming operations, they should be done in this method.
     // (Such as getAllKafkaPartitions() in KafkaRoutineLoad)
-    protected void preCheckNeedSchedule() throws UserException {
-
+    protected boolean preCheckNeedSchedule(boolean isNeedAutoSchedule) throws UserException {
+        return false;
     }
 
     protected void unprotectUpdateProgress() throws UserException {
     }
 
-    protected boolean unprotectNeedReschedule() throws UserException {
+    protected boolean unprotectNeedReschedule(boolean isNeedAutoSchedule) throws UserException {
+        return false;
+    }
+
+    protected boolean isNeedAutoSchedule() {
         return false;
     }