AsyncWorkItemHandler implements WorkItemHandler, Cacheable { protected long getProcessInstanceId(WorkItem workItem) { return ((WorkItemImpl) workItem).getProcessInstanceId(); } AsyncWorkItemHandler(ExecutorService executorService);  AsyncWorkItemHandler(ExecutorService executorService, String commandClass);  AsyncWorkItemHandler(Object executorService, String commandClass); @Override void executeWorkItem(WorkItem workItem, WorkItemManager manager); @Override void abortWorkItem(WorkItem workItem, WorkItemManager manager); @Override void close();  }
@Test(timeout=10000) public void testRunProcessWithAsyncHandlerRecordExecutionError() 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()); }
@Test(timeout=10000) public void testRunProcessWithAsyncHandlerSkipExecutionError() throws Exception { final NodeLeftCountDownProcessEventListener countDownListener = new NodeLeftCountDownProcessEventListener("Handling error", 1); ((ExecutorServiceImpl) executorService).setRetries(0); RuntimeEnvironment environment = RuntimeEnvironmentBuilder.Factory.get().newDefaultBuilder() .userGroupCallback(userGroupCallback) .addAsset(ResourceFactory.newClassPathResource("BPMN2-ScriptTaskErrorHandling.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; } @Override public List<ProcessEventListener> getProcessEventListeners( RuntimeEngine runtime) { List<ProcessEventListener> listeners = super.getProcessEventListeners(runtime); listeners.add(countDownListener); return listeners; } }) .get(); manager = RuntimeManagerFactory.Factory.get().newSingletonRuntimeManager(environment); assertNotNull(manager); RuntimeEngine runtime = manager.getRuntimeEngine(EmptyContext.get()); KieSession ksession = runtime.getKieSession(); assertNotNull(ksession); Map<String, Object> params = new HashMap<>(); params.put("command", "org.jbpm.executor.ThrowExceptionCommand"); ProcessInstance processInstance = ksession.startProcess("ScriptTask", params); assertEquals(ProcessInstance.STATE_ACTIVE, processInstance.getState()); countDownListener.waitTillCompleted(); List<RequestInfo> errorJobs = executorService.getInErrorRequests(new QueryContext()); assertEquals(1, errorJobs.size()); RequestInfo errorJob = errorJobs.get(0); assertEquals(errorJob.getProcessInstanceId().longValue(), processInstance.getId()); processInstance = runtime.getKieSession().getProcessInstance(processInstance.getId()); assertNull(processInstance); 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(0, errors.size()); }
@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); 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); 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()); }
@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) { } 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); 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); 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()); }
@SuppressWarnings("unchecked") @Test(timeout=20000) public void testRunProcessWithAsyncHandlerRecordExecutionErrorTaskAutoAck() 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(); TaskLifeCycleEventListener listener = new DefaultTaskEventListener(){ @Override public void afterTaskStartedEvent(TaskEvent event) { throw new TaskExecutionException("On purpose"); } }; try { ((EventService<TaskLifeCycleEventListener>)taskService).registerTaskEventListener(listener); taskService.start(taskId, "john"); fail("Start task should fail due to broken script"); } catch (Throwable e) { } 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("Task", error.getType()); assertEquals("UserTaskWithRollback", error.getProcessId()); assertEquals("Hello", 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); CommandContext ctx = new CommandContext(); ctx.setData("SingleRun", "true"); ctx.setData("EmfName", "org.jbpm.persistence.complete"); executorService.scheduleRequest(TaskAutoAckErrorCommand.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(); ((EventService<TaskLifeCycleEventListener>)taskService).removeTaskEventListener(listener); taskService.start(taskId, "john"); Map<String, Object> results = new HashMap<>(); results.put("output1", "ok"); taskService.complete(taskId, "john", results); manager.disposeRuntimeEngine(runtime); countDownListener.reset(1); executorService.scheduleRequest(TaskAutoAckErrorCommand.class.getName(), ctx); countDownListener.waitTillCompleted(); errors = errorStorage.list(0, 10); assertEquals(1, errors.size()); error = errors.get(0); assertNotNull(error); assertTrue(error.isAcknowledged()); }