feat(engine+rest): add job query filter for failed activity

related to CAM-11273

engine-rest/engine-rest/src/main/java/org/camunda/bpm/engine/rest/dto/history/HistoricJobLogQueryDto.java

@@ -31,7 +31,6 @@
 import org.camunda.bpm.engine.rest.dto.converter.BooleanConverter;
 import org.camunda.bpm.engine.rest.dto.converter.LongConverter;
 import org.camunda.bpm.engine.rest.dto.converter.StringArrayConverter;
-import org.camunda.bpm.engine.rest.dto.converter.StringConverter;
 import org.camunda.bpm.engine.rest.dto.converter.StringListConverter;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -86,6 +85,7 @@ public class HistoricJobLogQueryDto extends AbstractQueryDto<HistoricJobLogQuery
   protected String jobDefinitionType;
   protected String jobDefinitionConfiguration;
   protected String[] activityIds;
+  protected String[] failedActivityIds;
   protected String[] executionIds;
   protected String processInstanceId;
   protected String processDefinitionId;
@@ -142,6 +142,11 @@ public void setActivityIdIn(String[] activityIds) {
     this.activityIds = activityIds;
   }
 
+  @CamundaQueryParam(value="failedActivityIdIn", converter = StringArrayConverter.class)
+  public void setFailedActivityIdIn(String[] activityIds) {
+    this.failedActivityIds = activityIds;
+  }
+
   @CamundaQueryParam(value="executionIdIn", converter = StringArrayConverter.class)
   public void setExecutionIdIn(String[] executionIds) {
     this.executionIds = executionIds;
@@ -252,6 +257,10 @@ protected void applyFilters(HistoricJobLogQuery query) {
       query.activityIdIn(activityIds);
     }
 
+    if (failedActivityIds != null && failedActivityIds.length > 0) {
+      query.failedActivityIdIn(failedActivityIds);
+    }
+
     if (executionIds != null && executionIds.length > 0) {
       query.executionIdIn(executionIds);
     }

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

@@ -79,6 +79,7 @@ public class JobQueryDto extends AbstractQueryDto<JobQuery> {
   protected Boolean messages;
   protected Boolean withException;
   protected String exceptionMessage;
+  protected String failedActivityId;
   protected Boolean noRetriesLeft;
   protected Boolean active;
   protected Boolean suspended;
@@ -103,6 +104,11 @@ public void setActivityId(String activityId) {
     this.activityId = activityId;
   }
 
+  @CamundaQueryParam("failedActivityId")
+  public void setFailedActivityId(String activityId) {
+    this.failedActivityId = activityId;
+  }
+
   @CamundaQueryParam("jobId")
   public void setJobId(String jobId) {
     this.jobId = jobId;
@@ -257,7 +263,7 @@ void run(List<ConditionQueryParameterDto> dates) {
 
     abstract void setLowerThan(Date date);
   }
-  
+
   @Override
   protected void applyFilters(final JobQuery query) {
     if (activityId != null){
@@ -314,6 +320,10 @@ protected void applyFilters(final JobQuery query) {
       query.exceptionMessage(exceptionMessage);
     }
 
+    if (failedActivityId != null) {
+      query.failedActivityId(failedActivityId);
+    }
+
     if (TRUE.equals(noRetriesLeft)) {
       query.noRetriesLeft();
     }

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

@@ -409,6 +409,7 @@ private Map<String, Object> getCompleteParameters() {
     parameters.put("timers", MockProvider.EXAMPLE_TIMERS);
     parameters.put("withException", MockProvider.EXAMPLE_WITH_EXCEPTION);
     parameters.put("exceptionMessage", MockProvider.EXAMPLE_EXCEPTION_MESSAGE);
+    parameters.put("failedActivityId", MockProvider.EXAMPLE_JOB_FAILED_ACTIVITY_ID);
     parameters.put("noRetriesLeft", MockProvider.EXAMPLE_NO_RETRIES_LEFT);
     parameters.put("active", true);
     parameters.put("suspended", true);
@@ -445,6 +446,7 @@ private void verifyParameterQueryInvocations() {
     verify(mockQuery).timers();
     verify(mockQuery).withException();
     verify(mockQuery).exceptionMessage((String) parameters.get("exceptionMessage"));
+    verify(mockQuery).failedActivityId((String) parameters.get("failedActivityId"));
     verify(mockQuery).noRetriesLeft();
     verify(mockQuery).active();
     verify(mockQuery).suspended();

engine-rest/engine-rest/src/test/java/org/camunda/bpm/engine/rest/history/HistoricJobLogRestServiceQueryTest.java

@@ -598,6 +598,7 @@ public void testListParameters() {
     given()
       .queryParam("activityIdIn", anActId + "," + anotherActId)
       .queryParam("executionIdIn", anExecutionId + "," + anotherExecutionId)
+      .queryParam("failedActivityIdIn", anActId + "," + anotherActId)
     .then()
       .expect()
         .statusCode(Status.OK.getStatusCode())
@@ -606,6 +607,7 @@ public void testListParameters() {
 
     verify(mockedQuery).activityIdIn(anActId, anotherActId);
     verify(mockedQuery).executionIdIn(anExecutionId, anotherExecutionId);
+    verify(mockedQuery).failedActivityIdIn(anActId, anotherActId);
     verify(mockedQuery).list();
   }
 
@@ -620,6 +622,7 @@ public void testListParametersAsPost() {
     Map<String, List<String>> json = new HashMap<String, List<String>>();
     json.put("activityIdIn", Arrays.asList(anActId, anotherActId));
     json.put("executionIdIn", Arrays.asList(anExecutionId, anotherExecutionId));
+    json.put("failedActivityIdIn", Arrays.asList(anActId, anotherActId));
 
     given()
       .contentType(POST_JSON_CONTENT_TYPE)
@@ -632,6 +635,7 @@ public void testListParametersAsPost() {
 
     verify(mockedQuery).activityIdIn(anActId, anotherActId);
     verify(mockedQuery).executionIdIn(anExecutionId, anotherExecutionId);
+    verify(mockedQuery).failedActivityIdIn(anActId, anotherActId);
     verify(mockedQuery).list();
   }
 

engine/src/main/java/org/camunda/bpm/engine/history/HistoricJobLogQuery.java

@@ -45,6 +45,9 @@ public interface HistoricJobLogQuery extends Query<HistoricJobLogQuery, Historic
   /** Only select historic job log entries which are associated with one of the given activity ids. **/
   HistoricJobLogQuery activityIdIn(String... activityIds);
 
+  /** Only select historic job log entries which are associated with failures of one of the given activity ids. **/
+  HistoricJobLogQuery failedActivityIdIn(String... activityIds);
+
   /** Only select historic job log entries which are associated with one of the given execution ids. **/
   HistoricJobLogQuery executionIdIn(String... executionIds);
 

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

@@ -47,6 +47,7 @@ public class HistoricJobLogQueryImpl extends AbstractQuery<HistoricJobLogQuery,
   protected String jobDefinitionType;
   protected String jobDefinitionConfiguration;
   protected String[] activityIds;
+  protected String[] failedActivityIds;
   protected String[] executionIds;
   protected String processInstanceId;
   protected String processDefinitionId;
@@ -112,6 +113,14 @@ public HistoricJobLogQuery activityIdIn(String... activityIds) {
     return this;
   }
 
+  public HistoricJobLogQuery failedActivityIdIn(String... activityIds) {
+    List<String> activityIdList = CollectionUtil.asArrayList(activityIds);
+    ensureNotContainsNull("activityIds", activityIdList);
+    ensureNotContainsEmptyString("activityIds", activityIdList);
+    this.failedActivityIds = activityIds;
+    return this;
+  }
+
   public HistoricJobLogQuery executionIdIn(String... executionIds) {
     List<String> executionIdList = CollectionUtil.asArrayList(executionIds);
     ensureNotContainsNull("executionIds", executionIdList);
@@ -323,6 +332,10 @@ public String[] getActivityIds() {
     return activityIds;
   }
 
+  public String[] getFailedActivityIds() {
+    return failedActivityIds;
+  }
+
   public String[] getExecutionIds() {
     return executionIds;
   }

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

@@ -62,6 +62,7 @@ public class JobQueryImpl extends AbstractQuery<JobQuery, Job> implements JobQue
   protected Long priorityLowerThanOrEqual;
   protected boolean withException;
   protected String exceptionMessage;
+  protected String failedActivityId;
   protected boolean noRetriesLeft;
   protected SuspensionState suspensionState;
 
@@ -211,6 +212,12 @@ public JobQuery exceptionMessage(String exceptionMessage) {
     return this;
   }
 
+  public JobQuery failedActivityId(String activityId){
+    ensureNotNull("Provided activity id", activityId);
+    this.failedActivityId = activityId;
+    return this;
+  }
+
   public JobQuery noRetriesLeft() {
     noRetriesLeft = true;
     return this;

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

@@ -121,6 +121,9 @@ public interface JobQuery extends Query<JobQuery, Job> {
   /** Only select jobs that failed due to an exception with the given message. */
   JobQuery exceptionMessage(String exceptionMessage);
 
+  /** Only select jobs that failed due to an exception at an activity with the given id. **/
+  JobQuery failedActivityId(String activityId);
+
   /** Only select jobs which have no retries left */
   JobQuery noRetriesLeft();
 

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

@@ -326,6 +326,13 @@
       <if test="jobExceptionMessage != null">
         and RES.JOB_EXCEPTION_MSG_ = #{jobExceptionMessage}
       </if>
+      <if test="failedActivityIds != null &amp;&amp; failedActivityIds.length > 0">
+        and RES.FAILED_ACT_ID_ in
+        <foreach item="item" index="index" collection="failedActivityIds"
+                 open="(" separator="," close=")">
+          #{item}
+        </foreach>
+      </if>
       <if test="jobDefinitionId != null">
         and RES.JOB_DEF_ID_ = #{jobDefinitionId}
       </if>

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

@@ -360,6 +360,9 @@
       <if test="exceptionMessage">
       	and RES.EXCEPTION_MSG_ = #{exceptionMessage}
       </if>
+      <if test="failedActivityId != null">
+        and RES.FAILED_ACT_ID_ = #{failedActivityId}
+      </if>
       <if test="noRetriesLeft">
         and RES.RETRIES_ = 0
       </if>

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

@@ -575,6 +575,30 @@ public void testQueryByExceptionMessageNull() {
     }
   }
 
+  @Test
+  @Deployment(resources = {"org/camunda/bpm/engine/test/api/mgmt/ManagementServiceTest.testGetJobExceptionStacktrace.bpmn20.xml"})
+  public void testQueryByFailedActivityId(){
+    JobQuery query = managementService.createJobQuery().failedActivityId("theScriptTask");
+    verifyQueryResults(query, 0);
+
+    ProcessInstance processInstance = startProcessInstanceWithFailingJob();
+
+    query = managementService.createJobQuery().failedActivityId("theScriptTask");
+    verifyFailedJob(query, processInstance);
+  }
+
+  @Test
+  public void testQueryByInvalidFailedActivityId(){
+    JobQuery query = managementService.createJobQuery().failedActivityId("invalid");
+    verifyQueryResults(query, 0);
+
+    try {
+      managementService.createJobQuery().failedActivityId(null).list();
+      fail();
+    } catch (ProcessEngineException e) {}
+  }
+
+
   @Test
   public void testJobQueryWithExceptions() throws Throwable {
 

engine/src/test/java/org/camunda/bpm/engine/test/history/HistoricJobLogQueryTest.java

@@ -189,6 +189,54 @@ public void testQueryByInvalidJobExceptionMessage() {
     }
   }
 
+  @Deployment(resources = {"org/camunda/bpm/engine/test/history/HistoricJobLogTest.testAsyncContinuation.bpmn20.xml"})
+  @Test
+  public void testQueryByFailedActivityId() {
+    runtimeService.startProcessInstanceByKey("process");
+    String jobId = managementService.createJobQuery().singleResult().getId();
+    try {
+      managementService.executeJob(jobId);
+      fail("exception expected");
+    } catch (Exception e) {
+      // expected
+    }
+
+    HistoricJobLogQuery query = historyService.createHistoricJobLogQuery().failedActivityIdIn("serviceTask");
+
+    verifyQueryResults(query, 1);
+  }
+
+  @Test
+  public void testQueryByInvalidFailedActivityId() {
+    HistoricJobLogQuery query = historyService.createHistoricJobLogQuery().failedActivityIdIn("invalid");
+
+    verifyQueryResults(query, 0);
+
+    String[] nullValue = null;
+
+    try {
+      query.failedActivityIdIn(nullValue);
+      fail("exception expected");
+    } catch (Exception e) {
+    }
+
+    String[] activityIdsContainsNull = {"a", null, "b"};
+
+    try {
+      query.failedActivityIdIn(activityIdsContainsNull);
+      fail("exception expected");
+    } catch (Exception e) {
+    }
+
+    String[] activityIdsContainsEmptyString = {"a", "", "b"};
+
+    try {
+      query.failedActivityIdIn(activityIdsContainsEmptyString);
+      fail("exception expected");
+    } catch (Exception e) {
+    }
+  }
+
   @Deployment(resources = {"org/camunda/bpm/engine/test/history/HistoricJobLogTest.testAsyncContinuation.bpmn20.xml"})
   @Test
   public void testQueryByJobDefinitionId() {