for exp

Author: qq254963746

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

@@ -168,13 +168,15 @@ private RemotingCommand process(boolean receiveNewJob,
 
         // 判断是否接受新任务
         if (receiveNewJob) {
-            // 查看有没有其他可以执行的任务
-            JobPushRequest jobPushRequest = getNewJob(taskTrackerNodeGroup, taskTrackerIdentity);
-            // 返回 新的任务
-            return RemotingCommand.createResponseCommand(RemotingProtos
-                    .ResponseCode.SUCCESS.code(), jobPushRequest);
+            try {
+                // 查看有没有其他可以执行的任务
+                JobPushRequest jobPushRequest = getNewJob(taskTrackerNodeGroup, taskTrackerIdentity);
+                // 返回 新的任务
+                return RemotingCommand.createResponseCommand(RemotingProtos
+                        .ResponseCode.SUCCESS.code(), jobPushRequest);
+            } catch (Exception ignored) {
+            }
         }
-
         // 返回给 任务执行端
         return RemotingCommand.createResponseCommand(RemotingProtos
                 .ResponseCode.SUCCESS.code());
@@ -276,7 +278,9 @@ private JobPushRequest getNewJob(String taskTrackerNodeGroup, String taskTracker
         try {
             application.getExecutingJobQueue().add(jobPo);
         } catch (DuplicateJobException e) {
-            throw e;
+            LOGGER.warn(e.getMessage(), e);
+            application.getExecutableJobQueue().resume(jobPo);
+            return null;
         }
         application.getExecutableJobQueue().remove(jobPo.getTaskTrackerNodeGroup(), jobPo.getJobId());
 
@@ -304,9 +308,6 @@ private void finishProcess(List<TaskTrackerJobResult> results) {
         for (TaskTrackerJobResult result : results) {
 
             JobWrapper jobWrapper = result.getJobWrapper();
-            // 从正在执行的队列中移除 TODO 如果在这个时候down机了，数据丢失了
-            application.getExecutingJobQueue().remove(jobWrapper.getJobId());
-
             if (jobWrapper.getJob().isSchedule()) {
 
                 JobPo cronJobPo = application.getCronJobQueue().finish(jobWrapper.getJobId());
@@ -330,6 +331,9 @@ private void finishProcess(List<TaskTrackerJobResult> results) {
                     LOGGER.error(e.getMessage(), e);
                 }
             }
+            // 从正在执行的队列中移除
+            application.getExecutingJobQueue().remove(jobWrapper.getJobId());
+
         }
     }
 
@@ -347,9 +351,6 @@ private void retryProcess(List<TaskTrackerJobResult> results) {
             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,6 +390,10 @@ 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

@@ -171,7 +171,9 @@ public void operationComplete(ResponseFuture responseFuture) {
         try {
             application.getExecutingJobQueue().add(jobPo);
         } catch (DuplicateJobException e) {
-            throw e;
+            LOGGER.warn(e.getMessage(), e);
+            application.getExecutableJobQueue().resume(jobPo);
+            return PushResult.FAILED;
         }
         application.getExecutableJobQueue().remove(jobPo.getTaskTrackerNodeGroup(), jobPo.getJobId());
 

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

@@ -176,20 +176,20 @@ 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);
-            // 2. add to executable queue
+            // 1. add to executable queue
             try {
                 application.getExecutableJobQueue().add(jobPo);
             } catch (DuplicateJobException e) {
                 // ignore
-                LOGGER.error(e.getMessage(), e);
+                LOGGER.warn(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);