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

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."
+    },
+  "acquired": {
+    "type": "boolean",
+    "desc": "Only select jobs which are acquired, 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 acquired;
 
   protected List<ConditionQueryParameterDto> dueDates;
   protected List<ConditionQueryParameterDto> createTimes;
@@ -234,6 +235,11 @@ public void setIncludeJobsWithoutTenantId(Boolean includeJobsWithoutTenantId) {
     this.includeJobsWithoutTenantId = includeJobsWithoutTenantId;
   }
 
+  @CamundaQueryParam(value="acquired", converter = BooleanConverter.class)
+  public void setAcquired(Boolean acquired) {
+    this.acquired = acquired;
+  }
+
   @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(acquired)) {
+      query.acquired();
+    }
   }
 
   @Override

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

@@ -357,6 +357,17 @@ public void testMessagesParameter() {
     verify(mockQuery).list();
   }
 
+  @Test
+  public void testAcquiredParameter() {
+    Map<String, Object> parameters = new HashMap<>();
+    parameters.put("acquired", MockProvider.EXAMPLE_ACQUIRED);
+
+    given().queryParams(parameters).then().expect().statusCode(Status.OK.getStatusCode()).when().get(JOBS_RESOURCE_URL);
+
+    verify(mockQuery).acquired();
+    verify(mockQuery).list();
+  }
+
   @Test
   public void testMessagesTimersParameter() {
     Map<String, Object> parameters = new HashMap<>();
@@ -419,6 +430,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("acquired", MockProvider.EXAMPLE_ACQUIRED);
     return parameters;
   }
 
@@ -456,6 +468,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).acquired();
   }
 
   private void testDateParameters(DateParameters parameters) {

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

@@ -521,6 +521,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_ACQUIRED = 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 acquired;
 
   protected boolean isTenantIdSet = false;
   protected String[] tenantIds;
@@ -257,6 +258,11 @@ public JobQuery suspended() {
     return this;
   }
 
+  public JobQuery acquired() {
+    acquired = true;
+    return this;
+  }
+
   @Override
   protected boolean hasExcludingConditions() {
     return super.hasExcludingConditions()
@@ -402,5 +408,8 @@ public boolean isWithException() {
   public String getExceptionMessage() {
     return exceptionMessage;
   }
+  public boolean getAcquired() {
+    return acquired;
+  }
 
 }

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 currently being acquired,
+   * ie. lock expiration time is not null, lock expiration is in future and suspension state is 1
+   */
+  JobQuery acquired();
+
 }

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

@@ -419,6 +419,14 @@
       <if test="suspensionState != null">
         and RES.SUSPENSION_STATE_ = #{suspensionState.stateCode}
       </if>
+      <if test="acquired">
+        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 testQueryByAcquired() {
+    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().acquired().list();
+    assertEquals(list.size(), 1);
+    deleteJobInDatabase();
+
+    //given - lock expiration date in the past
+    lockExpDate.add(Calendar.MILLISECOND, -60000000);
+    createJobWithLockExpiration(lockExpDate.getTime());
+
+    list = managementService.createJobQuery().acquired().list();
+    assertEquals(list.size(), 0);
+
+    deleteJobInDatabase();
+  }
+
   //helper ////////////////////////////////////////////////////////////
 
   private void setRetries(final String processInstanceId, final int retries) {
@@ -886,6 +911,25 @@ public Void execute(CommandContext commandContext) {
     });
   }
 
+  private void createJobWithLockExpiration(Date lockDate) {
+    CommandExecutor commandExecutor = processEngineConfiguration.getCommandExecutorTxRequired();
+    commandExecutor.execute((Command<Void>) 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");

webapps/frontend/camunda-bpm-sdk-js/lib/api-client/resources/job.js

@@ -66,6 +66,8 @@ Job.get = function(id, done) {
  * @param  {String}   params.sorting.sortOrder      Sort the results in a given order. Values may be asc for ascending order or desc for descending order.
  * @param  {String}   [params.firstResult]          Pagination of results. Specifies the index of the first result to return.
  * @param  {String}   [params.maxResults]           Pagination of results. Specifies the maximum number of results to return. Will return less results if there are no more results left.
+ * @param  {Bool}     [params.acquired]             Select jobs which are acquired, ie. lock expiration time is not
+ *                                                  null,lock expiration is in future and suspension state is 1
  * @param  {Function} done
  */
 Job.list = function(params, done) {