for exp

Author: qq254963746

lts-jobtracker/src/main/java/com/lts/jobtracker/processor/JobFinishedProcessor.java

@@ -276,8 +276,7 @@ private JobPushRequest getNewJob(String taskTrackerNodeGroup, String taskTracker
         try {
             application.getExecutingJobQueue().add(jobPo);
         } catch (DuplicateJobException e) {
-            // ignore
-            LOGGER.error(e.getMessage(), e);
+            throw e;
         }
         application.getExecutableJobQueue().remove(jobPo.getTaskTrackerNodeGroup(), jobPo.getJobId());
 
@@ -305,34 +304,31 @@ private void finishProcess(List<TaskTrackerJobResult> results) {
         for (TaskTrackerJobResult result : results) {
 
             JobWrapper jobWrapper = result.getJobWrapper();
+            // 从正在执行的队列中移除 TODO 如果在这个时候down机了，数据丢失了
+            application.getExecutingJobQueue().remove(jobWrapper.getJobId());
 
-            try {
-                if (jobWrapper.getJob().isSchedule()) {
+            if (jobWrapper.getJob().isSchedule()) {
 
-                    JobPo cronJobPo = application.getCronJobQueue().finish(jobWrapper.getJobId());
-                    if (cronJobPo == null) {
-                        // 可能任务队列中改条记录被删除了
-                        return;
-                    }
-                    Date nextTriggerTime = CronExpressionUtils.getNextTriggerTime(cronJobPo.getCronExpression());
-                    if (nextTriggerTime == null) {
-                        application.getCronJobQueue().remove(jobWrapper.getJobId());
-                        return;
-                    }
-                    // 表示下次还要执行
-                    try {
-                        cronJobPo.setTaskTrackerIdentity(null);
-                        cronJobPo.setIsRunning(false);
-                        cronJobPo.setTriggerTime(nextTriggerTime.getTime());
-                        cronJobPo.setGmtModified(SystemClock.now());
-                        application.getExecutableJobQueue().add(cronJobPo);
-                    } catch (DuplicateJobException e) {
-                        LOGGER.error(e.getMessage(), e);
-                    }
+                JobPo cronJobPo = application.getCronJobQueue().finish(jobWrapper.getJobId());
+                if (cronJobPo == null) {
+                    // 可能任务队列中改条记录被删除了
+                    return;
+                }
+                Date nextTriggerTime = CronExpressionUtils.getNextTriggerTime(cronJobPo.getCronExpression());
+                if (nextTriggerTime == null) {
+                    application.getCronJobQueue().remove(jobWrapper.getJobId());
+                    return;
+                }
+                // 表示下次还要执行
+                try {
+                    cronJobPo.setTaskTrackerIdentity(null);
+                    cronJobPo.setIsRunning(false);
+                    cronJobPo.setTriggerTime(nextTriggerTime.getTime());
+                    cronJobPo.setGmtModified(SystemClock.now());
+                    application.getExecutableJobQueue().add(cronJobPo);
+                } catch (DuplicateJobException e) {
+                    LOGGER.error(e.getMessage(), e);
                 }
-            } finally {
-                // 移除
-                application.getExecutingJobQueue().remove(jobWrapper.getJobId());
             }
         }
     }
@@ -350,6 +346,10 @@ private void retryProcess(List<TaskTrackerJobResult> results) {
             // 1. 加入到重试队列
             JobPo jobPo = application.getExecutingJobQueue().get(jobWrapper.getJobId());
             if (jobPo != null) {
+
+                // 从正在执行的队列中移除 TODO 如果在这个时候down机了，数据丢失了
+                application.getExecutingJobQueue().remove(jobPo.getJobId());
+
                 // 重试次数+1
                 jobPo.setRetryTimes((jobPo.getRetryTimes() == null ? 0 : jobPo.getRetryTimes()) + 1);
                 Long nextRetryTriggerTime = DateUtils.addMinute(new Date(), jobPo.getRetryTimes()).getTime();
@@ -389,8 +389,6 @@ private void retryProcess(List<TaskTrackerJobResult> results) {
                         LOGGER.error(e.getMessage(), e);
                     }
                 }
-                // 从正在执行的队列中移除
-                application.getExecutingJobQueue().remove(jobPo.getJobId());
             }
         }
     }

lts-jobtracker/src/main/java/com/lts/jobtracker/support/JobPusher.java

@@ -167,15 +167,15 @@ public void operationComplete(ResponseFuture responseFuture) {
             application.getExecutableJobQueue().resume(jobPo);
             return PushResult.FAILED;
         }
+
         try {
             application.getExecutingJobQueue().add(jobPo);
         } catch (DuplicateJobException e) {
-            // ignore
-            LOGGER.error(e.getMessage(), e);
+            throw e;
         }
         application.getExecutableJobQueue().remove(jobPo.getTaskTrackerNodeGroup(), jobPo.getJobId());
-        // 记录日志
 
+        // 记录日志
         JobLogPo jobLogPo = JobDomainConverter.convertJobLog(jobPo);
         jobLogPo.setSuccess(true);
         jobLogPo.setLogType(LogType.SENT);

lts-jobtracker/src/main/java/com/lts/jobtracker/support/checker/ExecutingDeadJobChecker.java

@@ -175,18 +175,21 @@ private void fixDeadJob(List<JobPo> jobPos) {
 
     private void fixDeadJob(JobPo jobPo) {
         try {
+
+            // 1. remove from executing queue TODO 如果在这个时候down机了，数据丢失了
+            application.getExecutingJobQueue().remove(jobPo.getJobId());
+
             jobPo.setGmtModified(SystemClock.now());
             jobPo.setTaskTrackerIdentity(null);
             jobPo.setIsRunning(false);
-            // 1. add to executable queue
+            // 2. add to executable queue
             try {
                 application.getExecutableJobQueue().add(jobPo);
             } catch (DuplicateJobException e) {
                 // ignore
                 LOGGER.error(e.getMessage(), e);
             }
-            // 2. remove from executing queue
-            application.getExecutingJobQueue().remove(jobPo.getJobId());
+
             JobLogPo jobLogPo = JobDomainConverter.convertJobLog(jobPo);
             jobLogPo.setSuccess(true);
             jobLogPo.setLevel(Level.WARN);