@Test(timeout=20000)
public void testRunProcessWithAsyncHandlerRecordExecutionErrorProcessAutoAck() 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-UserTaskWithRollback.bpmn2"), ResourceType.BPMN2)
            .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("UserTaskWithRollback");
    assertEquals(ProcessInstance.STATE_ACTIVE, processInstance.getState());
    
    processInstance = runtime.getKieSession().getProcessInstance(processInstance.getId());
    assertNotNull(processInstance);
    
    manager.disposeRuntimeEngine(runtime);
    runtime = manager.getRuntimeEngine(EmptyContext.get());
    
    TaskService taskService = runtime.getTaskService();
    List<TaskSummary> tasks = taskService.getTasksAssignedAsPotentialOwner("john", "en-UK");
    assertEquals(1, tasks.size());
    
    long taskId = tasks.get(0).getId();
    
    taskService.start(taskId, "john");
    
    Map<String, Object> results = new HashMap<>();
    results.put("output1", "rollback");
    
    try {
        taskService.complete(taskId, "john", results);
        fail("Complete task should fail due to broken script");
    } catch (Throwable e) {
        // expected
    }
    
    manager.disposeRuntimeEngine(runtime);
    
    
    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("Process", error.getType());        
    assertEquals("UserTaskWithRollback", error.getProcessId());
    assertEquals("Script Task 1", 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(ProcessAutoAckErrorCommand.class.getName(), ctx);
            
    countDownListener.waitTillCompleted();
    
    errors = errorStorage.list(0, 10);
    assertEquals(1, errors.size());
    
    error = errors.get(0);
    assertNotNull(error);        
    assertFalse(error.isAcknowledged());
    
    runtime = manager.getRuntimeEngine(EmptyContext.get());
    tasks = taskService.getTasksAssignedAsPotentialOwner("john", "en-UK");
    assertEquals(1, tasks.size());
    
    taskId = tasks.get(0).getId();
    
    results = new HashMap<>();
    results.put("output1", "ok");
    
    taskService.complete(taskId, "john", results);
    manager.disposeRuntimeEngine(runtime);
    countDownListener.reset(1);
    // since task was completed auto ack should work
    executorService.scheduleRequest(ProcessAutoAckErrorCommand.class.getName(), ctx);
            
    countDownListener.waitTillCompleted();
    
    errors = errorStorage.list(0, 10);
    assertEquals(1, errors.size());
    
    error = errors.get(0);
    assertNotNull(error);        
    assertTrue(error.isAcknowledged());
    
}