@Test public void updateSlotStateRunningExternalIsRunning() throws Exception { SlotState slotState = new SlotState(slotId, SlotState.Status.RUNNING); ExternalStatus running = new MockExternalService.MockExternalStatusRunning(); when(externalService.getStatus(slotId, slotState.getExternalID())).thenReturn(running); scheduler.updateSlotState(wf, slotState, ScheduledTime.now(), mockedConnection); verifyNoMoreInteractions(mockedConnection); }
void updateSlotState(Workflow wf, SlotState slotState, ScheduledTime current, StateDatabaseConnection connection) throws Exception { SlotID slotID = slotState.getSlotID(); SlotState.Status status = slotState.getStatus(); if (status.equals(SlotState.Status.WAITING)) { if (callTrigger(wf, slotState, current, connection)) { LOGGER.info("Slot is ready: " + slotID); connection.putSlotState(slotState.transitionToReady()); } else if (isSlotTimedOut(slotState.getScheduledTime(), current, wf.getWaitTimeoutSeconds())) { LOGGER.info("Slot timed out waiting: " + slotID); connection.putSlotState(slotState.transitionToWaitTimeout()); } else { LOGGER.info("Waiting for slot: " + slotID); } } else if (status.equals(SlotState.Status.RUNNING)) { String externalID = slotState.getExternalID(); ExternalStatus xStatus = wf.getExternalService().getStatus(slotID, externalID); if (!xStatus.isRunning()) { if (xStatus.isSuccess()) { LOGGER.info("Slot successful: " + slotID + " external ID: " + externalID); connection.putSlotState(slotState.transitionToSuccess()); } else { if (slotState.getRetryCount() < wf.getMaxRetryCount()) { LOGGER.info("Slot failed, preparing for retry: " + slotID + " external ID: " + externalID); connection.putSlotState(slotState.transitionToRetry()); } else { LOGGER.info("Slot failed permanently: " + slotID + " external ID: " + externalID); connection.putSlotState(slotState.transitionToFailure()); } } } else { LOGGER.info("Slot still running: " + slotID + " external ID: " + externalID); } } }
Scheduler { void updateSlotState(Workflow wf, SlotState slotState, ScheduledTime current, StateDatabaseConnection connection) throws Exception { SlotID slotID = slotState.getSlotID(); SlotState.Status status = slotState.getStatus(); if (status.equals(SlotState.Status.WAITING)) { if (callTrigger(wf, slotState, current, connection)) { LOGGER.info("Slot is ready: " + slotID); connection.putSlotState(slotState.transitionToReady()); } else if (isSlotTimedOut(slotState.getScheduledTime(), current, wf.getWaitTimeoutSeconds())) { LOGGER.info("Slot timed out waiting: " + slotID); connection.putSlotState(slotState.transitionToWaitTimeout()); } else { LOGGER.info("Waiting for slot: " + slotID); } } else if (status.equals(SlotState.Status.RUNNING)) { String externalID = slotState.getExternalID(); ExternalStatus xStatus = wf.getExternalService().getStatus(slotID, externalID); if (!xStatus.isRunning()) { if (xStatus.isSuccess()) { LOGGER.info("Slot successful: " + slotID + " external ID: " + externalID); connection.putSlotState(slotState.transitionToSuccess()); } else { if (slotState.getRetryCount() < wf.getMaxRetryCount()) { LOGGER.info("Slot failed, preparing for retry: " + slotID + " external ID: " + externalID); connection.putSlotState(slotState.transitionToRetry()); } else { LOGGER.info("Slot failed permanently: " + slotID + " external ID: " + externalID); connection.putSlotState(slotState.transitionToFailure()); } } } else { LOGGER.info("Slot still running: " + slotID + " external ID: " + externalID); } } } }
Scheduler { void updateSlotState(Workflow wf, SlotState slotState, ScheduledTime current, StateDatabaseConnection connection) throws Exception { SlotID slotID = slotState.getSlotID(); SlotState.Status status = slotState.getStatus(); if (status.equals(SlotState.Status.WAITING)) { if (callTrigger(wf, slotState, current, connection)) { LOGGER.info("Slot is ready: " + slotID); connection.putSlotState(slotState.transitionToReady()); } else if (isSlotTimedOut(slotState.getScheduledTime(), current, wf.getWaitTimeoutSeconds())) { LOGGER.info("Slot timed out waiting: " + slotID); connection.putSlotState(slotState.transitionToWaitTimeout()); } else { LOGGER.info("Waiting for slot: " + slotID); } } else if (status.equals(SlotState.Status.RUNNING)) { String externalID = slotState.getExternalID(); ExternalStatus xStatus = wf.getExternalService().getStatus(slotID, externalID); if (!xStatus.isRunning()) { if (xStatus.isSuccess()) { LOGGER.info("Slot successful: " + slotID + " external ID: " + externalID); connection.putSlotState(slotState.transitionToSuccess()); } else { if (slotState.getRetryCount() < wf.getMaxRetryCount()) { LOGGER.info("Slot failed, preparing for retry: " + slotID + " external ID: " + externalID); connection.putSlotState(slotState.transitionToRetry()); } else { LOGGER.info("Slot failed permanently: " + slotID + " external ID: " + externalID); connection.putSlotState(slotState.transitionToFailure()); } } } else { LOGGER.info("Slot still running: " + slotID + " external ID: " + externalID); } } } Scheduler(WorkflowConfiguration configuration, int slidingWindowHours); }
Scheduler { void updateSlotState(Workflow wf, SlotState slotState, ScheduledTime current, StateDatabaseConnection connection) throws Exception { SlotID slotID = slotState.getSlotID(); SlotState.Status status = slotState.getStatus(); if (status.equals(SlotState.Status.WAITING)) { if (callTrigger(wf, slotState, current, connection)) { LOGGER.info("Slot is ready: " + slotID); connection.putSlotState(slotState.transitionToReady()); } else if (isSlotTimedOut(slotState.getScheduledTime(), current, wf.getWaitTimeoutSeconds())) { LOGGER.info("Slot timed out waiting: " + slotID); connection.putSlotState(slotState.transitionToWaitTimeout()); } else { LOGGER.info("Waiting for slot: " + slotID); } } else if (status.equals(SlotState.Status.RUNNING)) { String externalID = slotState.getExternalID(); ExternalStatus xStatus = wf.getExternalService().getStatus(slotID, externalID); if (!xStatus.isRunning()) { if (xStatus.isSuccess()) { LOGGER.info("Slot successful: " + slotID + " external ID: " + externalID); connection.putSlotState(slotState.transitionToSuccess()); } else { if (slotState.getRetryCount() < wf.getMaxRetryCount()) { LOGGER.info("Slot failed, preparing for retry: " + slotID + " external ID: " + externalID); connection.putSlotState(slotState.transitionToRetry()); } else { LOGGER.info("Slot failed permanently: " + slotID + " external ID: " + externalID); connection.putSlotState(slotState.transitionToFailure()); } } } else { LOGGER.info("Slot still running: " + slotID + " external ID: " + externalID); } } } Scheduler(WorkflowConfiguration configuration, int slidingWindowHours); void step(ScheduledTime current, StateDatabaseConnection connection); void step(ScheduledTime current, Set<WorkflowID> workflowIDs, StateDatabaseConnection connection); List<SlotState> getSlotStatesIncludingMarkedForRerun(Workflow wf, ScheduledTime current, ScheduledTime start, ScheduledTime end, StateDatabaseConnection connection); List<SlotState> getSlotStates(Workflow wf, ScheduledTime start, ScheduledTime end, StateDatabaseConnection connection); ScheduledTime getWorkflowStartTime(Workflow wf, ScheduledTime current); int getSlidingWindowHours(); WorkflowConfiguration getWorkflowConfiguration(); }
Scheduler { void updateSlotState(Workflow wf, SlotState slotState, ScheduledTime current, StateDatabaseConnection connection) throws Exception { SlotID slotID = slotState.getSlotID(); SlotState.Status status = slotState.getStatus(); if (status.equals(SlotState.Status.WAITING)) { if (callTrigger(wf, slotState, current, connection)) { LOGGER.info("Slot is ready: " + slotID); connection.putSlotState(slotState.transitionToReady()); } else if (isSlotTimedOut(slotState.getScheduledTime(), current, wf.getWaitTimeoutSeconds())) { LOGGER.info("Slot timed out waiting: " + slotID); connection.putSlotState(slotState.transitionToWaitTimeout()); } else { LOGGER.info("Waiting for slot: " + slotID); } } else if (status.equals(SlotState.Status.RUNNING)) { String externalID = slotState.getExternalID(); ExternalStatus xStatus = wf.getExternalService().getStatus(slotID, externalID); if (!xStatus.isRunning()) { if (xStatus.isSuccess()) { LOGGER.info("Slot successful: " + slotID + " external ID: " + externalID); connection.putSlotState(slotState.transitionToSuccess()); } else { if (slotState.getRetryCount() < wf.getMaxRetryCount()) { LOGGER.info("Slot failed, preparing for retry: " + slotID + " external ID: " + externalID); connection.putSlotState(slotState.transitionToRetry()); } else { LOGGER.info("Slot failed permanently: " + slotID + " external ID: " + externalID); connection.putSlotState(slotState.transitionToFailure()); } } } else { LOGGER.info("Slot still running: " + slotID + " external ID: " + externalID); } } } Scheduler(WorkflowConfiguration configuration, int slidingWindowHours); void step(ScheduledTime current, StateDatabaseConnection connection); void step(ScheduledTime current, Set<WorkflowID> workflowIDs, StateDatabaseConnection connection); List<SlotState> getSlotStatesIncludingMarkedForRerun(Workflow wf, ScheduledTime current, ScheduledTime start, ScheduledTime end, StateDatabaseConnection connection); List<SlotState> getSlotStates(Workflow wf, ScheduledTime start, ScheduledTime end, StateDatabaseConnection connection); ScheduledTime getWorkflowStartTime(Workflow wf, ScheduledTime current); int getSlidingWindowHours(); WorkflowConfiguration getWorkflowConfiguration();  }