FIX BUG

Author: qq254963746

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

@@ -328,12 +328,11 @@ private void finishProcess(List<TaskTrackerJobResult> results) {
                     cronJobPo.setGmtModified(SystemClock.now());
                     application.getExecutableJobQueue().add(cronJobPo);
                 } catch (DuplicateJobException e) {
-                    LOGGER.error(e.getMessage(), e);
+                    LOGGER.warn(e.getMessage(), e);
                 }
             }
             // 从正在执行的队列中移除
             application.getExecutingJobQueue().remove(jobWrapper.getJobId());
-
         }
     }
 

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

@@ -78,16 +78,25 @@ public void run() {
                         }
                     }
                     while (availableThreads > 0) {
-                        if(LOGGER.isDebugEnabled()){
+                        if (LOGGER.isDebugEnabled()) {
                             LOGGER.debug("taskTrackerNodeGroup:{}, taskTrackerIdentity:{} , availableThreads:{}", nodeGroup, identity, availableThreads);
                         }
                         // 推送任务
                         PushResult result = sendJob(remotingServer, taskTrackerNode);
-                        if (result == PushResult.SUCCESS) {
-                            availableThreads = taskTrackerNode.getAvailableThread().decrementAndGet();
-                            monitor.incPushJobNum();
-                        } else {
-                            break;
+                        switch (result) {
+                            case SUCCESS:
+                                availableThreads = taskTrackerNode.getAvailableThread().decrementAndGet();
+                                monitor.incPushJobNum();
+                                break;
+                            case FAILED:
+                                // 还是要继续发送
+                                break;
+                            case NO_JOB:
+                                // 没有任务了
+                                return;
+                            case SENT_ERROR:
+                                // TaskTracker链接失败
+                                return;
                         }
                     }
                 } catch (Exception e) {
@@ -100,7 +109,8 @@ public void run() {
     private enum PushResult {
         NO_JOB, // 没有任务可执行
         SUCCESS, //推送成功
-        FAILED      //推送失败
+        FAILED,      //推送失败
+        SENT_ERROR
     }
 
     /**
@@ -120,6 +130,17 @@ private PushResult sendJob(RemotingServerDelegate remotingServer, TaskTrackerNod
             return PushResult.NO_JOB;
         }
 
+        // IMPORTANT: 这里要先切换队列
+        try {
+            application.getExecutingJobQueue().add(jobPo);
+        } catch (DuplicateJobException e) {
+            LOGGER.warn(e.getMessage(), e);
+            application.getExecutableJobQueue().resume(jobPo);
+            return PushResult.FAILED;
+        }
+        application.getExecutableJobQueue().remove(jobPo.getTaskTrackerNodeGroup(), jobPo.getJobId());
+
+        // 发送给TaskTracker执行
         JobPushRequest body = application.getCommandBodyWrapper().wrapper(new JobPushRequest());
         body.setJobWrapper(JobDomainConverter.convert(jobPo));
         RemotingCommand commandRequest = RemotingCommand.createRequestCommand(JobProtos.RequestCode.PUSH_JOB.code(), body);
@@ -152,6 +173,7 @@ public void operationComplete(ResponseFuture responseFuture) {
 
         } catch (RemotingSendException e) {
             LOGGER.error(e.getMessage(), e);
+            return PushResult.SENT_ERROR;
         }
 
         try {
@@ -164,18 +186,21 @@ public void operationComplete(ResponseFuture responseFuture) {
             if (LOGGER.isDebugEnabled()) {
                 LOGGER.debug("Job push failed! nodeGroup=" + nodeGroup + ", identity=" + identity + ", job=" + jobPo);
             }
-            application.getExecutableJobQueue().resume(jobPo);
-            return PushResult.FAILED;
-        }
-
-        try {
-            application.getExecutingJobQueue().add(jobPo);
-        } catch (DuplicateJobException e) {
-            LOGGER.warn(e.getMessage(), e);
-            application.getExecutableJobQueue().resume(jobPo);
-            return PushResult.FAILED;
+            // 队列切回来
+            boolean needResume = true;
+            try {
+                jobPo.setIsRunning(true);
+                application.getExecutableJobQueue().add(jobPo);
+            } catch (DuplicateJobException e) {
+                LOGGER.warn(e.getMessage(), e);
+                needResume = false;
+            }
+            application.getExecutingJobQueue().remove(jobPo.getJobId());
+            if (needResume) {
+                application.getExecutableJobQueue().resume(jobPo);
+            }
+            return PushResult.SENT_ERROR;
         }
-        application.getExecutableJobQueue().remove(jobPo.getTaskTrackerNodeGroup(), jobPo.getJobId());
 
         // 记录日志
         JobLogPo jobLogPo = JobDomainConverter.convertJobLog(jobPo);