SimpleEventProcessor implements EventProcessor { @VisibleForTesting protected EventExecution execute(EventExecution eventExecution, Action action, Object payload) { try { String methodName = "executeEventAction"; String description = String.format("Executing action: %s for event: %s with messageId: %s with payload: %s", action.getAction(), eventExecution.getId(), eventExecution.getMessageId(), payload); logger.debug(description); Map<String, Object> output = new RetryUtil<Map<String, Object>>().retryOnException(() -> actionProcessor.execute(action, payload, eventExecution.getEvent(), eventExecution.getMessageId()), this::isTransientException, null, RETRY_COUNT, description, methodName); if (output != null) { eventExecution.getOutput().putAll(output); } eventExecution.setStatus(Status.COMPLETED); Monitors.recordEventExecutionSuccess(eventExecution.getEvent(), eventExecution.getName(), eventExecution.getAction().name()); } catch (RuntimeException e) { logger.error("Error executing action: {} for event: {} with messageId: {}", action.getAction(), eventExecution.getEvent(), eventExecution.getMessageId(), e); if (!isTransientException(e.getCause())) { eventExecution.setStatus(Status.FAILED); eventExecution.getOutput().put("exception", e.getMessage()); Monitors.recordEventExecutionError(eventExecution.getEvent(), eventExecution.getName(), eventExecution.getAction().name(), e.getClass().getSimpleName()); } } return eventExecution; } @Inject  SimpleEventProcessor(ExecutionService executionService,
                                MetadataService metadataService,
                                ActionProcessor actionProcessor,
                                EventQueues eventQueues,
                                JsonUtils jsonUtils,
                                Configuration configuration,
                                ObjectMapper objectMapper); Map<String, String> getQueues(); Map<String, Map<String, Long>> getQueueSizes();  }
@SuppressWarnings("unchecked") @Test public void testExecuteInvalidAction() { AtomicInteger executeInvoked = new AtomicInteger(0); doAnswer((Answer<Map<String, Object>>) invocation -> { executeInvoked.incrementAndGet(); throw new UnsupportedOperationException("error"); }).when(actionProcessor).execute(any(), any(), any(), any()); SimpleEventProcessor eventProcessor = new SimpleEventProcessor(executionService, metadataService, actionProcessor, eventQueues, jsonUtils, new TestConfiguration(), objectMapper); EventExecution eventExecution = new EventExecution("id", "messageId"); eventExecution.setName("handler"); eventExecution.setStatus(EventExecution.Status.IN_PROGRESS); eventExecution.setEvent("event"); Action action = new Action(); eventExecution.setAction(Type.start_workflow); eventProcessor.execute(eventExecution, action, "payload"); assertEquals(1, executeInvoked.get()); assertEquals(EventExecution.Status.FAILED, eventExecution.getStatus()); assertNotNull(eventExecution.getOutput().get("exception")); }
@Test public void testExecuteNonRetriableApplicationException() { AtomicInteger executeInvoked = new AtomicInteger(0); doAnswer((Answer<Map<String, Object>>) invocation -> { executeInvoked.incrementAndGet(); throw new ApplicationException(ApplicationException.Code.INVALID_INPUT, "some non-retriable error"); }).when(actionProcessor).execute(any(), any(), any(), any()); SimpleEventProcessor eventProcessor = new SimpleEventProcessor(executionService, metadataService, actionProcessor, eventQueues, jsonUtils, new TestConfiguration(), objectMapper); EventExecution eventExecution = new EventExecution("id", "messageId"); eventExecution.setStatus(EventExecution.Status.IN_PROGRESS); eventExecution.setEvent("event"); eventExecution.setName("handler"); Action action = new Action(); action.setAction(Type.start_workflow); eventExecution.setAction(Type.start_workflow); eventProcessor.execute(eventExecution, action, "payload"); assertEquals(1, executeInvoked.get()); assertEquals(EventExecution.Status.FAILED, eventExecution.getStatus()); assertNotNull(eventExecution.getOutput().get("exception")); }
@Test public void testExecuteRetriableApplicationException() { AtomicInteger executeInvoked = new AtomicInteger(0); doAnswer((Answer<Map<String, Object>>) invocation -> { executeInvoked.incrementAndGet(); throw new ApplicationException(ApplicationException.Code.BACKEND_ERROR, "some retriable error"); }).when(actionProcessor).execute(any(), any(), any(), any()); SimpleEventProcessor eventProcessor = new SimpleEventProcessor(executionService, metadataService, actionProcessor, eventQueues, jsonUtils, new TestConfiguration(), objectMapper); EventExecution eventExecution = new EventExecution("id", "messageId"); eventExecution.setStatus(EventExecution.Status.IN_PROGRESS); eventExecution.setEvent("event"); Action action = new Action(); action.setAction(Type.start_workflow); eventProcessor.execute(eventExecution, action, "payload"); assertEquals(3, executeInvoked.get()); assertNull(eventExecution.getOutput().get("exception")); }