@Test(timeout=20000)
public void testRunProcessWithAsyncHandlerRecordExecutionErrorAutoAck() throws Exception {

    CountDownAsyncJobListener countDownListener = new CountDownAsyncJobListener(1);
    ((ExecutorServiceImpl) executorService).addAsyncJobListener(countDownListener);
    ((ExecutorServiceImpl) executorService).setRetries(0);
    
    RuntimeEnvironment environment = RuntimeEnvironmentBuilder.Factory.get().newDefaultBuilder()
            .userGroupCallback(userGroupCallback)
            .addAsset(ResourceFactory.newClassPathResource("BPMN2-ScriptTask.bpmn2"), ResourceType.BPMN2)
            .registerableItemsFactory(new DefaultRegisterableItemsFactory() {
                @Override
                public Map<String, WorkItemHandler> getWorkItemHandlers(RuntimeEngine runtime) {
                    Map<String, WorkItemHandler> handlers = super.getWorkItemHandlers(runtime);
                    handlers.put("async", new AsyncWorkItemHandler(executorService, "org.jbpm.executor.ThrowExceptionCommand"));
                    return handlers;
                }
          
            })
            .get();
    
    manager = RuntimeManagerFactory.Factory.get().newSingletonRuntimeManager(environment); 
    assertNotNull(manager);
    
    RuntimeEngine runtime = manager.getRuntimeEngine(EmptyContext.get());
    KieSession ksession = runtime.getKieSession();
    assertNotNull(ksession);       
    
    ProcessInstance processInstance = ksession.startProcess("ScriptTask");
    assertEquals(ProcessInstance.STATE_ACTIVE, processInstance.getState());
    
    countDownListener.waitTillCompleted();
    
    processInstance = runtime.getKieSession().getProcessInstance(processInstance.getId());
    assertNotNull(processInstance);
    
    List<RequestInfo> errorJobs = executorService.getInErrorRequests(new QueryContext());
    assertEquals(1, errorJobs.size());
    
    RequestInfo errorJob = errorJobs.get(0);
    assertEquals(errorJob.getProcessInstanceId().longValue(), processInstance.getId());
    
    ExecutionErrorManager errorManager = ((AbstractRuntimeManager) manager).getExecutionErrorManager();
    assertNotNull("ErrorManager is null", errorManager);
    ExecutionErrorStorage errorStorage = errorManager.getStorage();
    assertNotNull("ErrorStorage is null", errorStorage);
    
    List<ExecutionError> errors = errorStorage.list(0, 10);
    assertEquals(1, errors.size());
    
    ExecutionError error = errors.get(0);
    assertNotNull(error);
    assertEquals("Job", error.getType());
    assertEquals(errorJob.getId(), error.getJobId());
    assertEquals("ScriptTask", error.getProcessId());
    assertEquals("", error.getActivityName());
    assertEquals(manager.getIdentifier(), error.getDeploymentId());
    assertNotNull(error.getError());
    assertNotNull(error.getErrorMessage());
    assertNotNull(error.getActivityId());
    assertNotNull(error.getProcessInstanceId());
    
    assertNull(error.getAcknowledgedAt());
    assertNull(error.getAcknowledgedBy());
    assertFalse(error.isAcknowledged());
    
    countDownListener.reset(1);
    // first run should not ack the job as it's in error state
    CommandContext ctx = new CommandContext();
    ctx.setData("SingleRun", "true");
    ctx.setData("EmfName", "org.jbpm.persistence.complete");
    executorService.scheduleRequest(JobAutoAckErrorCommand.class.getName(), ctx);
            
    countDownListener.waitTillCompleted();
    
    errors = errorStorage.list(0, 10);
    assertEquals(1, errors.size());
    
    error = errors.get(0);
    assertNotNull(error);        
    assertFalse(error.isAcknowledged());
            
    executorService.cancelRequest(errorJob.getId());        
    countDownListener.reset(1);
    // since job was canceled auto ack should work
    executorService.scheduleRequest(JobAutoAckErrorCommand.class.getName(), ctx);
            
    countDownListener.waitTillCompleted();
    
    errors = errorStorage.list(0, 10);
    assertEquals(1, errors.size());
    
    error = errors.get(0);
    assertNotNull(error);        
    assertTrue(error.isAcknowledged());
    
}