[FIX BUG]ExecutingDeadJobChecker停止

lts-core/src/main/java/com/github/ltsopensource/cmd/HttpCmdAcceptor.java

@@ -11,6 +11,7 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.RejectedExecutionHandler;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
@@ -32,7 +33,15 @@ public HttpCmdAcceptor(ServerSocket serverSocket, HttpCmdContext context) {
         this.executorService = new ThreadPoolExecutor(Constants.AVAILABLE_PROCESSOR,
                 Constants.AVAILABLE_PROCESSOR,
                 0L, TimeUnit.MILLISECONDS,
-                new LinkedBlockingQueue<Runnable>(100), new ThreadPoolExecutor.DiscardPolicy());
+                new LinkedBlockingQueue<Runnable>(100), new RejectedExecutionHandler() {
+            @Override
+            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
+                if (r instanceof HttpCmdExecutor) {
+                    HttpCmdExecutor httpCmdExecutor = (HttpCmdExecutor)r;
+                    httpCmdExecutor.abort();
+                }
+            }
+        });
     }
 
     public void start() {

lts-core/src/main/java/com/github/ltsopensource/cmd/HttpCmdExecutor.java

@@ -51,6 +51,11 @@ public void run() {
         } catch (Throwable t) {
             LOGGER.error("Error When Execute Command", t);
             sendError(HTTP_INTERNALERROR, JSON.toJSONString(HttpCmdResponse.newResponse(false, "Error:" + t.getMessage())), false);
+        } finally {
+            try {
+                socket.close();
+            } catch (IOException e) {
+            }
         }
     }
 
@@ -111,6 +116,12 @@ private HttpCmdRequest parseRequest() throws Exception {
         return resolveRequest(uri, params);
     }
 
+    public void abort() {
+        try {
+            this.socket.close();
+        } catch (IOException e) {
+        }
+    }
 
     protected static HttpCmdRequest resolveRequest(String uri, Properties params) {
 

lts-jobtracker/src/main/java/com/github/ltsopensource/jobtracker/support/checker/ExecutableDeadJobChecker.java

@@ -29,7 +29,7 @@ public class ExecutableDeadJobChecker {
     // 1 分钟还锁着的，说明是有问题的
     private static final long MAX_TIME_OUT = 60 * 1000;
 
-    private final ScheduledExecutorService FIXED_EXECUTOR_SERVICE = Executors.newScheduledThreadPool(1, new NamedThreadFactory("LTS-ExecutableJobQueue-Fix-Executor", true));
+    private ScheduledExecutorService FIXED_EXECUTOR_SERVICE;
 
     private JobTrackerAppContext appContext;
 
@@ -43,6 +43,7 @@ public ExecutableDeadJobChecker(JobTrackerAppContext appContext) {
     public void start() {
         try {
             if (start.compareAndSet(false, true)) {
+                FIXED_EXECUTOR_SERVICE = Executors.newScheduledThreadPool(1, new NamedThreadFactory("LTS-ExecutableJobQueue-Fix-Executor", true));
                 scheduledFuture = FIXED_EXECUTOR_SERVICE.scheduleWithFixedDelay(new Runnable() {
                     @Override
                     public void run() {

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

@@ -50,7 +50,7 @@ public class ExecutingDeadJobChecker {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(ExecutingDeadJobChecker.class);
 
-    private final ScheduledExecutorService FIXED_EXECUTOR_SERVICE = Executors.newScheduledThreadPool(1, new NamedThreadFactory("LTS-ExecutingJobQueue-Fix-Executor", true));
+    private ScheduledExecutorService FIXED_EXECUTOR_SERVICE;
 
     private JobTrackerAppContext appContext;
     private JobTrackerMStatReporter stat;
@@ -73,6 +73,7 @@ public void start() {
                     fixCheckPeriodSeconds = 5 * 60;
                 }
 
+                FIXED_EXECUTOR_SERVICE = Executors.newScheduledThreadPool(1, new NamedThreadFactory("LTS-ExecutingJobQueue-Fix-Executor", true));
                 scheduledFuture = FIXED_EXECUTOR_SERVICE.scheduleWithFixedDelay(new Runnable() {
                     @Override
                     public void run() {

lts-jobtracker/src/main/java/com/github/ltsopensource/jobtracker/support/checker/FeedbackJobSendChecker.java

@@ -29,7 +29,7 @@ public class FeedbackJobSendChecker {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(FeedbackJobSendChecker.class);
 
-    private ScheduledExecutorService RETRY_EXECUTOR_SERVICE = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("LTS-FeedbackJobSend-Executor", true));
+    private ScheduledExecutorService RETRY_EXECUTOR_SERVICE;
     private ScheduledFuture<?> scheduledFuture;
     private AtomicBoolean start = new AtomicBoolean(false);
     private ClientNotifier clientNotifier;
@@ -68,6 +68,7 @@ public void handleFailed(List<JobRunResultWrapper> jobResults) {
     public void start() {
         try {
             if (start.compareAndSet(false, true)) {
+                RETRY_EXECUTOR_SERVICE = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("LTS-FeedbackJobSend-Executor", true));
                 scheduledFuture = RETRY_EXECUTOR_SERVICE.scheduleWithFixedDelay(new Runner()
                         , 30, 30, TimeUnit.SECONDS);
             }