WorkFlowExecutionController { @Transactional @SelectDataSource(type = DataSourceType.READ_WRITE, storage = Storage.SHARDED) protected Set<State> cancelPath(String stateMachineId, EventData eventData) { Set<State> executableStates = new HashSet<>(); StateMachine stateMachine = stateMachinesDAO.findById(stateMachineId); Context context = new RAMContext(System.currentTimeMillis(), null, stateMachine); Map<String, Event.EventStatus> eventStatusMap = eventsDAO.getAllEventsNameAndStatus(stateMachine.getId(), true); Queue<String> cancelledEvents = new LinkedList<>(); cancelledEvents.add(eventData.getName()); while (!cancelledEvents.isEmpty()) { String eventName = cancelledEvents.poll(); eventsDAO.markEventAsCancelled(stateMachine.getId(), eventName); eventStatusMap.put(eventName, Event.EventStatus.cancelled); final Set<State> dependantStates = context.getDependantStates(eventName); for (State state : dependantStates) { List<String> dependencies = state.getDependencies(); boolean allCancelled = true; boolean allMet = true; for (String dependency : dependencies) { if (eventStatusMap.get(dependency) != Event.EventStatus.cancelled) { allCancelled = false; } if (!(eventStatusMap.get(dependency) == Event.EventStatus.cancelled || eventStatusMap.get(dependency) == Event.EventStatus.triggered)) { allMet = false; } } if (allCancelled) { statesDAO.updateStatus(state.getStateMachineId(), state.getId(), Status.cancelled); auditDAO.create(state.getStateMachineId(), new AuditRecord(stateMachine.getId(), state.getId(), state.getAttemptedNoOfRetries(), Status.cancelled, null, null)); EventDefinition eventDefinition = null; if (state.getOutputEvent() != null) { try { eventDefinition = objectMapper.readValue(state.getOutputEvent(), EventDefinition.class); } catch (IOException ex) { throw new RuntimeException("Error occurred while deserializing task outputEvent for stateMachineInstanceId: " + stateMachine.getId() + " taskId: " + state.getId()); } cancelledEvents.add(eventDefinition.getName()); } } else if (allMet) { executableStates.add(state); } } } return executableStates; } @Inject  WorkFlowExecutionController(EventsDAO eventsDAO, StateMachinesDAO stateMachinesDAO,
                                       StatesDAO statesDAO, AuditDAO auditDAO, ExecutionNodeTaskDispatcher executionNodeTaskDispatcher,
                                       RedriverRegistry redriverRegistry, MetricsClient metricsClient,
                                       ClientElbPersistenceService clientElbPersistenceService); Set<State> initAndStart(StateMachine stateMachine); void updateTaskStatusAndPostEvent(StateMachine stateMachine, EventAndExecutionData eventAndExecutionData); void updateTaskStatusAndHandlePathCancellation(String stateMachineId, EventAndExecutionData eventAndExecutionData); void handlePathCancellation(String stateMachineId, EventData eventData); Set<State> postEvent(EventData eventData, String stateMachineInstanceId); @Transactional @SelectDataSource(type = DataSourceType.READ_WRITE, storage = Storage.SHARDED) Event persistEvent(String stateMachineInstanceId, EventData eventData); Set<State> processEvent(Event event, String stateMachineInstanceId); @Transactional @SelectDataSource(type = DataSourceType.READ_WRITE, storage = Storage.SHARDED) void updateTaskStatus(String machineId, Long stateId, ExecutionUpdateData executionUpdateData); void updateExecutionStatus(String stateMachineId, Long taskId, Status status, long retryCount, long currentRetryCount, String errorMessage, boolean deleteFromRedriver); @Transactional @SelectDataSource(type = DataSourceType.READ_WRITE, storage = Storage.SHARDED) void updateEventData(String machineId, EventData eventData); void unsidelineState(String stateMachineId, Long stateId); void incrementExecutionRetries(String stateMachineId, Long taskId); void redriveTask(String machineId, Long taskId); void cancelWorkflow(StateMachine stateMachine); static String getRouterName(String taskName);  }
@Test public void testCancelPath_shouldCancelPathTillJoinNode() throws Exception { HashMap<String, Event.EventStatus> eventStatusHashMap = new HashMap<String, Event.EventStatus>() {{ put("event1", Event.EventStatus.triggered); put("event2", Event.EventStatus.triggered); put("event3", Event.EventStatus.pending); put("event4", Event.EventStatus.pending); put("event5", Event.EventStatus.pending); }}; String outputEvent1 = null; String outputEvent2 = null; String outputEvent3 = null; String outputEvent4 = null; String outputEvent5 = null; String outputEvent6 = null; try { outputEvent1 = objectMapper.writeValueAsString(new EventDefinition("event1", "SomeEvent.class")); outputEvent2 = objectMapper.writeValueAsString(new EventDefinition("event2", "SomeEvent.class")); outputEvent3 = objectMapper.writeValueAsString(new EventDefinition("event3", "SomeEvent.class")); outputEvent4 = objectMapper.writeValueAsString(new EventDefinition("event4", "SomeEvent.class")); outputEvent5 = objectMapper.writeValueAsString(new EventDefinition("event5", "SomeEvent.class")); outputEvent6 = objectMapper.writeValueAsString(new EventDefinition("event6", "SomeEvent.class")); } catch (Exception e) { e.printStackTrace(); } State state1 = new State(1L, "state1", null, null, null, null, new ArrayList<>(), 0L, 1000L, outputEvent1, Status.initialized, null, 0L, "state-machine-cancel-path", 1L); State state2 = new State(1L, "state2", null, null, null, null, new ArrayList<String>() {{ add("event1"); }}, 0L, 1000L, outputEvent2, Status.initialized, null, 0L, "state-machine-cancel-path", 2L); State state3 = new State(1L, "state3", null, null, null, null, new ArrayList<String>() {{ add("event2"); add("event6"); }}, 0L, 1000L, null, Status.initialized, null, 0L, "state-machine-cancel-path", 3L); State state4 = new State(1L, "state4", null, null, null, null, new ArrayList<String>() {{ add("event1"); }}, 0L, 1000L, outputEvent3, Status.initialized, null, 0L, "state-machine-cancel-path", 4L); State state5 = new State(1L, "state5", null, null, null, null, new ArrayList<String>() {{ add("event3"); }}, 0L, 1000L, outputEvent4, Status.initialized, null, 0L, "state-machine-cancel-path", 5L); State state6 = new State(1L, "state6", null, null, null, null, new ArrayList<String>() {{ add("event3"); }}, 0L, 1000L, outputEvent5, Status.initialized, null, 0L, "state-machine-cancel-path", 6L); State state7 = new State(1L, "state7", null, null, null, null, new ArrayList<String>() {{ add("event4"); add("event5"); }}, 0L, 1000L, outputEvent6, Status.initialized, null, 0L, "state-machine-cancel-path", 7L); Set<State> states = new HashSet<State>() {{ add(state1); add(state2); add(state3); add(state4); add(state5); add(state6); add(state7); }}; StateMachine stateMachine = new StateMachine("state-machine-cancel-path", 1L, "state_machine_1", null, states, "client_elb_id_1"); EventData testEventData = new EventData("event3", null, null, "runtime", true); when(eventsDAO.getAllEventsNameAndStatus("state-machine-cancel-path", true)).thenReturn(eventStatusHashMap); when(stateMachinesDAO.findById("state-machine-cancel-path")).thenReturn(stateMachine); Set<State> executableStates = workFlowExecutionController.cancelPath(stateMachine.getId(), testEventData); assertThat(executableStates.size()).isEqualTo(1); assertThat(executableStates.contains("state3")); verify(eventsDAO).markEventAsCancelled("state-machine-cancel-path", "event3"); verify(eventsDAO).markEventAsCancelled("state-machine-cancel-path", "event4"); verify(eventsDAO).markEventAsCancelled("state-machine-cancel-path", "event5"); verify(eventsDAO).markEventAsCancelled("state-machine-cancel-path", "event6"); verify(statesDAO).updateStatus("state-machine-cancel-path", 5L, Status.cancelled); verify(statesDAO).updateStatus("state-machine-cancel-path", 6L, Status.cancelled); verify(statesDAO).updateStatus("state-machine-cancel-path", 7L, Status.cancelled); }