Add TaskTracker FailStore close config

Author: qq254963746

lts-core/src/main/java/com/github/ltsopensource/core/constant/ExtConfig.java

@@ -226,4 +226,7 @@ public interface ExtConfig {
     String JOB_TRACKER_PUSH_BATCH_SIZE = "lts.job.tracker.push.batch.size";
 
     String TASK_TRACKER_BIZ_LOGGER_FAIL_STORE_CLOSE = "lts.task.tracker.biz.logger.failstore.close";
+
+    String TASK_TRACKER_JOB_RESULT_FAIL_STORE_CLOSE = "lts.task.tracker.job.result.failstore.close";
+
 }

lts-tasktracker/src/main/java/com/github/ltsopensource/tasktracker/logger/BizLoggerImpl.java

@@ -49,27 +49,31 @@ public BizLoggerImpl(Level level, final RemotingClientDelegate remotingClient, T
         }
         this.appContext = appContext;
         this.remotingClient = remotingClient;
-        this.retryScheduler = new RetryScheduler<BizLog>(BizLogger.class.getSimpleName(), appContext, FailStorePathBuilder.getBizLoggerPath(appContext)) {
-            @Override
-            protected boolean isRemotingEnable() {
-                return remotingClient.isServerEnable();
-            }
 
-            @Override
-            protected boolean retry(List<BizLog> list) {
-                return sendBizLog(list);
-            }
-        };
         if (isEnableBizLoggerFailStore()) {
+
+            this.retryScheduler = new RetryScheduler<BizLog>(BizLogger.class.getSimpleName(), appContext, FailStorePathBuilder.getBizLoggerPath(appContext)) {
+                @Override
+                protected boolean isRemotingEnable() {
+                    return remotingClient.isServerEnable();
+                }
+
+                @Override
+                protected boolean retry(List<BizLog> list) {
+                    return sendBizLog(list);
+                }
+            };
+
             this.retryScheduler.start();
+
+            NodeShutdownHook.registerHook(appContext, this.getClass().getName(), new Callable() {
+                @Override
+                public void call() throws Exception {
+                    retryScheduler.stop();
+                }
+            });
         }
 
-        NodeShutdownHook.registerHook(appContext, this.getClass().getName(), new Callable() {
-            @Override
-            public void call() throws Exception {
-                retryScheduler.stop();
-            }
-        });
     }
 
     @Override
@@ -112,7 +116,11 @@ private void sendMsg(String msg, Level level) {
         requestBody.setBizLogs(Collections.singletonList(bizLog));
 
         if (!remotingClient.isServerEnable()) {
-            retryScheduler.inSchedule(StringUtils.generateUUID(), bizLog);
+            if(isEnableBizLoggerFailStore()){
+                retryScheduler.inSchedule(StringUtils.generateUUID(), bizLog);
+            }else{
+                logger.error("Send Biz Logger to JobTracker Error, server is down, bizLog={}", JSON.toJSONString(bizLog));
+            }
             return;
         }
 

lts-tasktracker/src/main/java/com/github/ltsopensource/tasktracker/processor/JobPushProcessor.java

@@ -3,6 +3,7 @@
 import com.github.ltsopensource.core.commons.utils.Callable;
 import com.github.ltsopensource.core.commons.utils.CollectionUtils;
 import com.github.ltsopensource.core.constant.Constants;
+import com.github.ltsopensource.core.constant.ExtConfig;
 import com.github.ltsopensource.core.domain.JobMeta;
 import com.github.ltsopensource.core.domain.JobRunResult;
 import com.github.ltsopensource.core.exception.JobTrackerNotFoundException;
@@ -50,29 +51,32 @@ public class JobPushProcessor extends AbstractProcessor {
     protected JobPushProcessor(TaskTrackerAppContext appContext) {
         super(appContext);
         this.remotingClient = appContext.getRemotingClient();
-        retryScheduler = new RetryScheduler<JobRunResult>(JobPushProcessor.class.getSimpleName(), appContext,
-                FailStorePathBuilder.getJobFeedbackPath(appContext), 3) {
-            @Override
-            protected boolean isRemotingEnable() {
-                return remotingClient.isServerEnable();
-            }
-
-            @Override
-            protected boolean retry(List<JobRunResult> results) {
-                return retrySendJobResults(results);
-            }
-        };
-        retryScheduler.start();
-
         // 线程安全的
         jobRunnerCallback = new JobRunnerCallback();
 
-        NodeShutdownHook.registerHook(appContext, this.getClass().getName(), new Callable() {
-            @Override
-            public void call() throws Exception {
-                retryScheduler.stop();
-            }
-        });
+
+        if (isEnableFailStore()) {
+            retryScheduler = new RetryScheduler<JobRunResult>(JobPushProcessor.class.getSimpleName(), appContext,
+                    FailStorePathBuilder.getJobFeedbackPath(appContext), 3) {
+                @Override
+                protected boolean isRemotingEnable() {
+                    return remotingClient.isServerEnable();
+                }
+
+                @Override
+                protected boolean retry(List<JobRunResult> results) {
+                    return retrySendJobResults(results);
+                }
+            };
+            retryScheduler.start();
+
+            NodeShutdownHook.registerHook(appContext, this.getClass().getName(), new Callable() {
+                @Override
+                public void call() throws Exception {
+                    retryScheduler.stop();
+                }
+            });
+        }
     }
 
     @Override
@@ -152,9 +156,15 @@ public void operationComplete(ResponseFuture responseFuture) {
                                     LOGGER.info("Job feedback failed, save local files。{}", jobRunResult);
                                 }
                                 try {
-                                    retryScheduler.inSchedule(
-                                            jobRunResult.getJobMeta().getJobId().concat("_") + SystemClock.now(),
-                                            jobRunResult);
+                                    if (isEnableFailStore()) {
+                                        retryScheduler.inSchedule(
+                                                jobRunResult.getJobMeta().getJobId().concat("_") + SystemClock.now(),
+                                                jobRunResult);
+                                    } else {
+                                        LOGGER.error("Send Job Result to JobTracker Error, code={}, jobRunResult={}",
+                                                commandResponse != null ? commandResponse.getCode() : null, JSON.toJSONString(jobRunResult));
+                                    }
+
                                 } catch (Exception e) {
                                     LOGGER.error("Job feedback failed", e);
                                 }
@@ -173,9 +183,14 @@ public void operationComplete(ResponseFuture responseFuture) {
             } catch (JobTrackerNotFoundException e) {
                 try {
                     LOGGER.warn("No job tracker available! save local files.");
-                    retryScheduler.inSchedule(
-                            jobRunResult.getJobMeta().getJobId().concat("_") + SystemClock.now(),
-                            jobRunResult);
+
+                    if (isEnableFailStore()) {
+                        retryScheduler.inSchedule(
+                                jobRunResult.getJobMeta().getJobId().concat("_") + SystemClock.now(),
+                                jobRunResult);
+                    } else {
+                        LOGGER.error("Send Job Result to JobTracker Error, server is down, jobRunResult={}", JSON.toJSONString(jobRunResult));
+                    }
                 } catch (Exception e1) {
                     LOGGER.error("Save files failed, {}", jobRunResult.getJobMeta(), e1);
                 }
@@ -185,6 +200,10 @@ public void operationComplete(ResponseFuture responseFuture) {
         }
     }
 
+    private boolean isEnableFailStore() {
+        return !appContext.getConfig().getParameter(ExtConfig.TASK_TRACKER_JOB_RESULT_FAIL_STORE_CLOSE, false);
+    }
+
     /**
      * 发送JobResults
      */