feat(engine, engine-rest-core) Add a query criteria to retrieve all executing jobs

related to: camunda#4470

Author: prajwolbhandari1

engine-rest/engine-rest-openapi/src/main/templates/lib/commons/job-definition-query-params.ftl

@@ -15,7 +15,7 @@
 <#elseif requestMethod == "POST">
   <#assign listSeparator = "">
 </#if>
-            
+
 <#assign params = {
   "jobDefinitionId": {
     "type": "string",

engine-rest/engine-rest-openapi/src/main/templates/lib/commons/job-query-params.ftl

@@ -163,5 +163,11 @@
     "type": "boolean",
     "desc": "Include jobs which belong to no tenant. Can be used in combination with `tenantIdIn`.
              Value may only be `true`, as `false` is the default behavior."
+    },
+  "executing": {
+    "type": "boolean",
+    "desc": "Only select jobs which are executing, i.e.,  lock expiration date is not null, lock expiration
+             date is in future and suspension state is 1. Value may only be `true`, as `false` is the default
+             behavior."
   }
 }>

engine-rest/engine-rest/src/main/java/org/camunda/bpm/engine/rest/dto/runtime/JobQueryDto.java

@@ -94,6 +94,7 @@ public class JobQueryDto extends AbstractQueryDto<JobQuery> {
   protected List<String> tenantIds;
   protected Boolean withoutTenantId;
   protected Boolean includeJobsWithoutTenantId;
+  protected Boolean executing;
 
   protected List<ConditionQueryParameterDto> dueDates;
   protected List<ConditionQueryParameterDto> createTimes;
@@ -234,6 +235,11 @@ public void setIncludeJobsWithoutTenantId(Boolean includeJobsWithoutTenantId) {
     this.includeJobsWithoutTenantId = includeJobsWithoutTenantId;
   }
 
+  @CamundaQueryParam(value="executing", converter = BooleanConverter.class)
+  public void setExecuting(Boolean executing) {
+    this.executing = executing;
+  }
+
   @Override
   protected boolean isValidSortByValue(String value) {
     return VALID_SORT_BY_VALUES.contains(value);
@@ -418,6 +424,9 @@ String fieldName() {
     if (TRUE.equals(includeJobsWithoutTenantId)) {
       query.includeJobsWithoutTenantId();
     }
+    if (TRUE.equals(executing)) {
+      query.executing();
+    }
   }
 
   @Override

engine-rest/engine-rest/src/test/java/org/camunda/bpm/engine/rest/JobRestServiceQueryTest.java

@@ -359,6 +359,17 @@ public void testMessagesParameter() {
     verify(mockQuery).list();
   }
 
+  @Test
+  public void testExecutingParameter() {
+    Map<String, Object> parameters = new HashMap<>();
+    parameters.put("executing", MockProvider.EXAMPLE_EXECUTING);
+
+    given().queryParams(parameters).then().expect().statusCode(Status.OK.getStatusCode()).when().get(JOBS_RESOURCE_URL);
+
+    verify(mockQuery).executing();
+    verify(mockQuery).list();
+  }
+
   @Test
   public void testMessagesTimersParameter() {
     Map<String, Object> parameters = new HashMap<>();
@@ -421,6 +432,7 @@ private Map<String, Object> getCompleteParameters() {
     parameters.put("priorityLowerThanOrEquals", JOB_QUERY_MAX_PRIORITY);
     parameters.put("priorityHigherThanOrEquals", JOB_QUERY_MIN_PRIORITY);
     parameters.put("jobDefinitionId", MockProvider.EXAMPLE_JOB_DEFINITION_ID);
+    parameters.put("executing", MockProvider.EXAMPLE_EXECUTING);
     return parameters;
   }
 
@@ -458,6 +470,7 @@ private void verifyParameterQueryInvocations() {
     verify(mockQuery).priorityLowerThanOrEquals(JOB_QUERY_MAX_PRIORITY);
     verify(mockQuery).priorityHigherThanOrEquals(JOB_QUERY_MIN_PRIORITY);
     verify(mockQuery).jobDefinitionId(MockProvider.EXAMPLE_JOB_DEFINITION_ID);
+    verify(mockQuery).executing();
   }
 
   private void testDateParameters(DateParameters parameters) {

engine-rest/engine-rest/src/test/java/org/camunda/bpm/engine/rest/helper/MockProvider.java

@@ -530,6 +530,8 @@ public abstract class MockProvider {
   public static final String EXAMPLE_AUTHORIZATION_ID = "someAuthorizationId";
   public static final int EXAMPLE_AUTHORIZATION_TYPE = 0;
   public static final String EXAMPLE_AUTHORIZATION_TYPE_STRING = "0";
+  public static final Boolean EXAMPLE_EXECUTING = true;
+
 
   // process applications
   public static final String EXAMPLE_PROCESS_APPLICATION_NAME = "aProcessApplication";

engine/src/main/java/org/camunda/bpm/engine/impl/JobQueryImpl.java

@@ -71,6 +71,7 @@ public class JobQueryImpl extends AbstractQuery<JobQuery, Job> implements JobQue
   protected String failedActivityId;
   protected boolean noRetriesLeft;
   protected SuspensionState suspensionState;
+  protected boolean executing;
 
   protected boolean isTenantIdSet = false;
   protected String[] tenantIds;
@@ -257,6 +258,11 @@ public JobQuery suspended() {
     return this;
   }
 
+  public JobQuery executing() {
+    executing = true;
+    return this;
+  }
+
   @Override
   protected boolean hasExcludingConditions() {
     return super.hasExcludingConditions()
@@ -402,5 +408,8 @@ public boolean isWithException() {
   public String getExceptionMessage() {
     return exceptionMessage;
   }
+  public boolean getExecuting() {
+    return executing;
+  }
 
 }

engine/src/main/java/org/camunda/bpm/engine/runtime/JobQuery.java

@@ -191,4 +191,10 @@ public interface JobQuery extends Query<JobQuery, Job> {
    */
   JobQuery orderByTenantId();
 
+  /**
+   * Only select jobs that are executing,
+   * ie. lock expiration time is not null, lock expiration is in future and suspension state is 1
+   */
+  JobQuery executing();
+
 }

engine/src/main/resources/org/camunda/bpm/engine/impl/mapping/entity/Job.xml

@@ -420,6 +420,14 @@
       <if test="suspensionState != null">
         and RES.SUSPENSION_STATE_ = #{suspensionState.stateCode}
       </if>
+      <if test="executing">
+        AND
+        RES.SUSPENSION_STATE_ = 1
+        AND
+        RES.LOCK_EXP_TIME_ IS NOT NULL
+        AND
+        RES.LOCK_EXP_TIME_  > #{now, jdbcType=TIMESTAMP}
+      </if>
       <if test="isTenantIdSet">
         <if test="tenantIds != null &amp;&amp; tenantIds.length > 0">
           and ( RES.TENANT_ID_ in

engine/src/test/java/org/camunda/bpm/engine/test/api/mgmt/JobQueryTest.java

@@ -871,6 +871,31 @@ public void testQueryInvalidSortingUsage() {
     }
   }
 
+  @Test
+  public void testQueryByExecuting() {
+    Calendar lockExpDate = Calendar.getInstance();
+    //given - lock expiration date in future
+    lockExpDate.add(Calendar.MILLISECOND, 30000000);
+
+    createJobWithLockExpiration(lockExpDate.getTime());
+
+    Job job = managementService.createJobQuery().jobId(timerEntity.getId()).singleResult();
+    assertNotNull(job);
+
+    List<Job> list = managementService.createJobQuery().executing().list();
+    assertEquals(list.size(), 1);
+    deleteJobInDatabase();
+
+    //given - lock expiration date in the past
+    lockExpDate.add(Calendar.MILLISECOND, -60000000);
+    createJobWithLockExpiration(lockExpDate.getTime());
+
+    list = managementService.createJobQuery().executing().list();
+    assertEquals(list.size(), 0);
+
+    deleteJobInDatabase();
+  }
+
   //helper ////////////////////////////////////////////////////////////
 
   private void setRetries(final String processInstanceId, final int retries) {
@@ -886,6 +911,27 @@ public Void execute(CommandContext commandContext) {
     });
   }
 
+  private void createJobWithLockExpiration(Date lockDate) {
+    CommandExecutor commandExecutor = processEngineConfiguration.getCommandExecutorTxRequired();
+    commandExecutor.execute(new Command<Void>() {
+      public Void execute(CommandContext commandContext) {
+        JobManager jobManager = commandContext.getJobManager();
+        timerEntity = new TimerEntity();
+        timerEntity.setLockOwner(UUID.randomUUID().toString());
+        timerEntity.setDuedate(new Date());
+        timerEntity.setRetries(0);
+        timerEntity.setLockExpirationTime(lockDate);
+
+        jobManager.insert(timerEntity);
+
+        assertNotNull(timerEntity.getId());
+
+        return null;
+
+      }
+    });
+  }
+
   private ProcessInstance startProcessInstanceWithFailingJob() {
     // start a process with a failing job
     ProcessInstance processInstance = runtimeService.startProcessInstanceByKey("exceptionInJobExecution");