add enqueue retry times and add item priority

Author: yunheli

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

@@ -35,6 +35,12 @@ 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;
+
     @Autowired
     private PlatformQueue<PriorityMessage> cmdCallbackQueue;
 
@@ -58,20 +64,35 @@ public void onQueueItem(PriorityMessage message) {
             jobService.callback(item);
         } catch (NotFoundException notFoundException) {
 
+            // detect retry times is reach the limit or not
+            detectRetryTimes(item);
+
             // re-enqueue cmd callback if job not found since transaction problem
-            reEnqueueJobCallback(item, 1000);
+            reEnqueueJobCallback(item, REQUEUE_DELAY_TIME);
 
         } catch (Throwable throwable) {
             LOGGER.traceMarker("onQueueItem", String.format("exception - %s", throwable));
         }
     }
 
+    private void detectRetryTimes(CmdCallbackQueueItem item) {
+        if (item.getRetryTimes() > RETRY_TIMES) {
+            throw new NotFoundException(String.format("retry times has reach the limit"));
+        }
+    }
+
     private void reEnqueueJobCallback(CmdCallbackQueueItem item, long wait) {
         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);
+
         jobService.enterQueue(item);
     }
 }

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

@@ -33,12 +33,27 @@ public class CmdCallbackQueueItem extends Jsonable {
 
     private Integer retryTimes = 0;
 
+    // priority default 1
+    private Integer priority = 1;
+
     public CmdCallbackQueueItem(BigInteger jobId, Cmd cmd) {
         this.jobId = jobId;
         this.cmd = cmd;
         this.path = cmd.getExtra();
     }
 
+    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/JobServiceImpl.java

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