change retryTimes dec and set priority param

Author: yunheli

platform-api/src/main/java/com/flow/platform/api/consumer/CmdCallbackQueueConsumer.java

@@ -36,9 +36,6 @@ public class CmdCallbackQueueConsumer implements QueueListener<PriorityMessage>
 
     private final static Logger LOGGER = new Logger(CmdCallbackQueueConsumer.class);
 
-    // retry time set 5 times
-    private final static int RETRY_TIMES = 5;
-
     // requeue 1 s
     private final static int REQUEUE_DELAY_TIME = 1000;
 
@@ -69,31 +66,31 @@ public void onQueueItem(PriorityMessage message) {
             detectRetryTimes(item);
 
             // re-enqueue cmd callback if job not found since transaction problem
-            reEnqueueJobCallback(item, REQUEUE_DELAY_TIME);
+            reEnqueueJobCallback(item, REQUEUE_DELAY_TIME, message.getMessageProperties().getPriority());
 
         } catch (Throwable throwable) {
             LOGGER.traceMarker("onQueueItem", String.format("exception - %s", throwable));
         }
     }
 
     private void detectRetryTimes(CmdCallbackQueueItem item) {
-        if (item.getRetryTimes() > RETRY_TIMES) {
-            throw new FlowException(String.format("retry times has reach the limit"));
+        if (item.getRetryTimes() <= 0) {
+            throw new FlowException("retry times has reach the limit");
         }
     }
 
-    private void reEnqueueJobCallback(CmdCallbackQueueItem item, long wait) {
+    private void reEnqueueJobCallback(CmdCallbackQueueItem item, long wait, int priority) {
         try {
             Thread.sleep(wait);
         } catch (Throwable ignore) {
         }
 
-        // set item priority priority inc
-        item.setPriority(item.getPriority() + 1);
-
         // set retry times
-        item.setRetryTimes(item.getRetryTimes() + 1);
+        item.setRetryTimes(item.getRetryTimes() - 1);
+
+        //priority inc 1
+        priority = priority + 1;
 
-        jobService.enterQueue(item);
+        jobService.enterQueue(item, priority);
     }
 }

platform-api/src/main/java/com/flow/platform/api/controller/CmdWebhookController.java

@@ -64,7 +64,7 @@ public void execute(@RequestBody Cmd cmd, @RequestParam String identifier) {
                 cmd.getStatus(),
                 cmd.getId(),
                 cmd.getCmdResult());
-            jobService.enterQueue(new CmdCallbackQueueItem(jobId, cmd));
+            jobService.enterQueue(new CmdCallbackQueueItem(jobId, cmd), 1);
         } catch (NumberFormatException warn) {
             LOGGER.warn("Invalid job id format");
         }

platform-api/src/main/java/com/flow/platform/api/domain/CmdCallbackQueueItem.java

@@ -31,10 +31,8 @@ public class CmdCallbackQueueItem extends Jsonable {
 
     private final Cmd cmd;
 
-    private Integer retryTimes = 1;
-
-    // priority default 1
-    private Integer priority = 1;
+    // default retry times 5
+    private Integer retryTimes = 5;
 
     public CmdCallbackQueueItem(BigInteger jobId, Cmd cmd) {
         this.jobId = jobId;
@@ -46,14 +44,6 @@ public void setRetryTimes(Integer retryTimes) {
         this.retryTimes = retryTimes;
     }
 
-    public Integer getPriority() {
-        return priority;
-    }
-
-    public void setPriority(Integer priority) {
-        this.priority = priority;
-    }
-
     public BigInteger getJobId() {
         return jobId;
     }

platform-api/src/main/java/com/flow/platform/api/service/job/JobService.java

@@ -110,7 +110,7 @@ void createWithYmlLoad(String path,
     /**
      * Enqueue cmd callback item
      */
-    void enterQueue(CmdCallbackQueueItem cmdQueueItem);
+    void enterQueue(CmdCallbackQueueItem cmdQueueItem, int priority);
 
     /**
      * stop job

platform-api/src/main/java/com/flow/platform/api/service/job/JobServiceImpl.java

@@ -462,8 +462,8 @@ private void onRunShellCallback(String path, Cmd cmd, Job job) {
     }
 
     @Override
-    public void enterQueue(CmdCallbackQueueItem cmdQueueItem) {
-        cmdCallbackQueue.enqueue(PriorityMessage.create(cmdQueueItem.toBytes(), cmdQueueItem.getPriority()));
+    public void enterQueue(CmdCallbackQueueItem cmdQueueItem, int priority) {
+        cmdCallbackQueue.enqueue(PriorityMessage.create(cmdQueueItem.toBytes(), priority));
     }
 
     @Override

platform-api/src/main/java/com/flow/platform/api/service/node/NodeCrontabService.java

@@ -44,4 +44,9 @@ public interface NodeCrontabService extends ContextEvent {
      * List current triggers
      */
     List<Trigger> triggers();
+
+    /**
+     * Clean triggers
+     */
+    void cleanTriggers();
 }

platform-api/src/main/java/com/flow/platform/api/service/node/NodeCrontabServiceImpl.java

@@ -165,4 +165,13 @@ public List<Trigger> triggers() {
             throw new IllegalStatusException(e.getMessage());
         }
     }
+
+    @Override
+    public void cleanTriggers() {
+        try {
+            quartzScheduler.getTriggersOfJob(nodeCrontabDetail.getKey()).clear();
+        } catch (SchedulerException e) {
+            throw new IllegalStatusException(e.getMessage());
+        }
+    }
 }

platform-api/src/test/java/com/flow/platform/api/test/service/JobServiceTest.java

@@ -325,10 +325,13 @@ public void should_cmd_enqueue_limit_times_success() throws InterruptedException
         });
 
         // when: enter queue one not found job id
-        jobService.enterQueue(new CmdCallbackQueueItem(CommonUtil.randomId(), cmd));
+        jobService.enterQueue(new CmdCallbackQueueItem(CommonUtil.randomId(), cmd), 1);
         countDownLatch.await(6, TimeUnit.SECONDS);
 
         // then: should try 5 times
-        Assert.assertEquals(5, atomicInteger.get());
+        Assert.assertEquals(1, atomicInteger.get());
+
+        // then: cmdCallbackQueue size should 0
+        Assert.assertEquals(0, cmdCallbackQueue.size());
     }
 }

platform-api/src/test/java/com/flow/platform/api/test/service/NodeCrontabServiceTest.java

@@ -42,6 +42,7 @@ public class NodeCrontabServiceTest extends TestBase {
     @Before
     public void before() {
         stubDemo();
+        flowCrontabService.cleanTriggers();
     }
 
     @Test