@Test public void schedulingOnlySubsetOfWorkflowsWorks() throws Exception { WorkflowID wfID1 = new WorkflowID("wf1"); WorkflowID wfID2 = new WorkflowID("wf2"); Schedule sch1 = makeHourlySchedule(); SchedulingStrategy str1 = makeSerialSchedulingStrategy(); Trigger tr1 = makeAlwaysTrigger(); MockExternalService srv1 = new MockExternalService(new MockExternalService.MockExternalStatusRunning()); int maxRetryCount = 0; Workflow wf1 = new Workflow(wfID1, sch1, str1, tr1, srv1, maxRetryCount, Workflow.DEFAULT_START_TIME, Workflow.DEFAULT_WAIT_TIMEOUT_SECONDS, emptyWorkflowInfo); Workflow wf2 = new Workflow(wfID2, sch1, str1, tr1, srv1, maxRetryCount, Workflow.DEFAULT_START_TIME, Workflow.DEFAULT_WAIT_TIMEOUT_SECONDS, emptyWorkflowInfo); WorkflowConfiguration cfg = new WorkflowConfiguration(); cfg.addWorkflow(wf1); cfg.addWorkflow(wf2); MemoryStateDatabase db = new MemoryStateDatabase(); SlotID id1 = new SlotID(wfID1, new ScheduledTime("2013-11-27T20:00Z")); SlotID id2 = new SlotID(wfID2, new ScheduledTime("2013-11-27T20:00Z")); SlotState slot1 = new SlotState(id1, SlotState.Status.WAITING); SlotState slot2 = new SlotState(id2, SlotState.Status.WAITING); connection.putSlotState(slot1); connection.putSlotState(slot2); Set<WorkflowID> subset = new HashSet<>(); subset.add(wfID1); int slidingWindowHours = 3; DateTime current = DateTime.parse("2013-11-27T22:01Z"); Scheduler sched = new Scheduler(cfg, slidingWindowHours); sched.step(new ScheduledTime(current), subset, connection); SlotState slot1After = connection.getSlotState(id1); Assert.assertEquals(SlotState.Status.READY, slot1After.getStatus()); SlotState slot2After = connection.getSlotState(id2); Assert.assertEquals(SlotState.Status.WAITING, slot2After.getStatus()); }
public void step(ScheduledTime current, StateDatabaseConnection connection) throws Exception { step(current, Collections.<WorkflowID>emptySet(), connection); }
Scheduler { public void step(ScheduledTime current, StateDatabaseConnection connection) throws Exception { step(current, Collections.<WorkflowID>emptySet(), connection); } }
Scheduler { public void step(ScheduledTime current, StateDatabaseConnection connection) throws Exception { step(current, Collections.<WorkflowID>emptySet(), connection); } Scheduler(WorkflowConfiguration configuration, int slidingWindowHours); }
Scheduler { public void step(ScheduledTime current, StateDatabaseConnection connection) throws Exception { step(current, Collections.<WorkflowID>emptySet(), connection); } 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 { public void step(ScheduledTime current, StateDatabaseConnection connection) throws Exception { step(current, Collections.<WorkflowID>emptySet(), connection); } 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();  }