加入LTS自带业务重试功能

Author: 胡贵

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

@@ -287,6 +287,7 @@ private void finishProcess(List<TaskTrackerJobResult> results) {
                 } else {
                     // 表示下次还要执行
                     try {
+                        cronJobPo.setTaskTrackerIdentity(null);
                         cronJobPo.setIsRunning(false);
                         cronJobPo.setTriggerTime(nextTriggerTime.getTime());
                         cronJobPo.setGmtModified(System.currentTimeMillis());
@@ -316,8 +317,6 @@ private void retryProcess(List<TaskTrackerJobResult> results) {
                 // 重试次数+1
                 jobPo.setRetryTimes((jobPo.getRetryTimes() == null ? 0 : jobPo.getRetryTimes()) + 1);
                 Long nextRetryTriggerTime = DateUtils.addMinute(new Date(), jobPo.getRetryTimes()).getTime();
-                // 延迟重试时间就等于重试次数(分钟)
-                jobPo.setTriggerTime(nextRetryTriggerTime);
 
                 boolean needAdd = true;
 
@@ -329,6 +328,7 @@ private void retryProcess(List<TaskTrackerJobResult> results) {
                         if (nextTriggerTime != null && nextTriggerTime.getTime() < nextRetryTriggerTime) {
                             // 表示下次还要执行, 并且下次执行时间比下次重试时间要早, 那么不重试，直接使用下次的执行时间
                             try {
+                                cronJobPo.setTaskTrackerIdentity(null);
                                 cronJobPo.setIsRunning(false);
                                 cronJobPo.setTriggerTime(nextTriggerTime.getTime());
                                 cronJobPo.setGmtModified(System.currentTimeMillis());
@@ -345,6 +345,9 @@ private void retryProcess(List<TaskTrackerJobResult> results) {
                 if (needAdd) {
                     // 加入到队列, 重试
                     jobPo.setIsRunning(false);
+                    jobPo.setTaskTrackerIdentity(null);
+                    // 延迟重试时间就等于重试次数(分钟)
+                    jobPo.setTriggerTime(nextRetryTriggerTime);
                     application.getExecutableJobQueue().add(jobPo);
                 }
                 // 从正在执行的队列中移除

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

@@ -2,6 +2,7 @@
 
 import com.lts.biz.logger.domain.JobLogPo;
 import com.lts.biz.logger.domain.LogType;
+import com.lts.core.commons.utils.DateUtils;
 import com.lts.core.constant.Constants;
 import com.lts.core.constant.Level;
 import com.lts.core.domain.Job;
@@ -171,6 +172,7 @@ public void operationComplete(ResponseFuture responseFuture) {
         JobLogPo jobLogPo = JobDomainConverter.convertJobLog(jobPo);
         jobLogPo.setSuccess(true);
         jobLogPo.setLogType(LogType.SENT);
+        jobLogPo.setLogTime(DateUtils.currentTimeMillis());
         jobLogPo.setLevel(Level.INFO);
         application.getJobLogger().log(jobLogPo);
 

lts-logger/lts-logger-mysql/src/main/java/com/lts/biz/logger/mysql/MysqlJobLogger.java

@@ -39,8 +39,8 @@ public MysqlJobLogger(Config config) {
         insertSQL = "INSERT INTO `lts_job_log_po` (`log_time`,`gmt_created`, `log_type`, `success`, `msg`" +
                 ",`task_tracker_identity`, `level`, `task_id`, `job_id`" +
                 ", `priority`, `submit_node_group`, `task_tracker_node_group`, `ext_params`, `need_feedback`" +
-                ", `cron_expression`, `trigger_time`)" +
-                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
+                ", `cron_expression`, `trigger_time`, `retry_times`)" +
+                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
     }
 
     @Override
@@ -65,7 +65,8 @@ public void log(JobLogPo jobLogPo) {
                     JSONUtils.toJSONString(jobLogPo.getExtParams()),
                     jobLogPo.isNeedFeedback(),
                     jobLogPo.getCronExpression(),
-                    jobLogPo.getTriggerTime()
+                    jobLogPo.getTriggerTime(),
+                    jobLogPo.getRetryTimes()
             );
         } catch (SQLException e) {
             throw new JobLogException(e.getMessage(), e);
@@ -80,17 +81,17 @@ public void log(List<JobLogPo> jobLogPos) {
         String prefixSQL = "INSERT INTO `lts_job_log_po` ( `log_time`, `gmt_created`, `log_type`, `success`, `msg`" +
                 ",`task_tracker_identity`, `level`, `task_id`, `job_id`" +
                 ", `priority`, `submit_node_group`, `task_tracker_node_group`, `ext_params`, `need_feedback`" +
-                ", `cron_expression`, `trigger_time`) VALUES ";
+                ", `cron_expression`, `trigger_time`, `retry_times`) VALUES ";
         int size = jobLogPos.size();
         for (int i = 0; i < size; i++) {
             if (i == size - 1) {
-                prefixSQL += "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
+                prefixSQL += "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
             } else {
-                prefixSQL += "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?),";
+                prefixSQL += "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?),";
             }
         }
 
-        Object[][] params = new Object[size][16];
+        Object[][] params = new Object[size][17];
         int index = 0;
         for (JobLogPo jobLogPo : jobLogPos) {
             int i = index++;
@@ -110,6 +111,7 @@ public void log(List<JobLogPo> jobLogPos) {
             params[i][13] = jobLogPo.isNeedFeedback();
             params[i][14] = jobLogPo.getCronExpression();
             params[i][15] = jobLogPo.getTriggerTime();
+            params[i][16] = jobLogPo.getRetryTimes();
         }
 
         try {
@@ -142,6 +144,7 @@ public List<JobLogPo> handle(ResultSet rs) throws SQLException {
                 jobLogPo.setNeedFeedback(rs.getBoolean("need_feedback"));
                 jobLogPo.setCronExpression(rs.getString("cron_expression"));
                 jobLogPo.setTriggerTime(rs.getLong("trigger_time"));
+                jobLogPo.setRetryTimes(rs.getInt("retry_times"));
                 result.add(jobLogPo);
             }
             return result;

lts-logger/lts-logger-mysql/src/main/resources/sql/lts_job_log_po.sql

@@ -14,9 +14,10 @@ CREATE TABLE IF NOT EXISTS `lts_job_log_po` (
   `submit_node_group` varchar(64) DEFAULT NULL COMMENT '提交节点group',
   `task_tracker_node_group` varchar(64) DEFAULT NULL COMMENT '执行节点group',
   `ext_params` text COMMENT '额外参数',
-  `needFeedback` tinyint(4) DEFAULT NULL COMMENT '是否需要反馈',
+  `need_feedback` tinyint(4) DEFAULT NULL COMMENT '是否需要反馈',
   `cron_expression` varchar(32) DEFAULT NULL COMMENT 'cron表达式',
   `trigger_time` bigint(20) DEFAULT NULL COMMENT '触发时间',
+  `retry_times` int(11) DEFAULT NULL COMMENT '重试次数',
   PRIMARY KEY (`id`),
   KEY `log_time` (`log_time`),
   KEY `task_id_task_tracker_node_group` (`task_id`,`task_tracker_node_group`)

lts-queue/lts-queue-mysql/src/main/resources/sql/lts_cron_job_queue.sql

@@ -16,5 +16,5 @@ CREATE TABLE IF NOT EXISTS `{tableName}` (
   `trigger_time` bigint(20) DEFAULT NULL COMMENT '下一次执行时间',
   PRIMARY KEY (`id`),
   UNIQUE KEY `idx_job_id` (`job_id`),
-  UNIQUE KEY `idx_taskId_taskTrackerNodeGroup` (`task_id`, `task_tracker_node_group`),
+  UNIQUE KEY `idx_taskId_taskTrackerNodeGroup` (`task_id`, `task_tracker_node_group`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

lts-queue/lts-queue-mysql/src/main/resources/sql/lts_node_group_store.sql

@@ -1,6 +1,6 @@
 CREATE TABLE IF NOT EXISTS `{tableName}` (
   `node_type` varchar(16) NOT NULL DEFAULT '' COMMENT '节点类型',
   `name` varchar(64) NOT NULL DEFAULT '' COMMENT '名字',
-  `gmt_created` timestamp NULL DEFAULT NULL COMMENT '创建时间',
+  `gmt_created` bigint(20) NULL DEFAULT NULL COMMENT '创建时间',
   PRIMARY KEY (`node_type`,`name`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8;