BackfillTriggerManager { void tick() { try (Scope ignored = tracer.spanBuilder("Styx.BackfillTriggerManager.tick") .setRecordEvents(true) .setSampler(Samplers.alwaysSample()) .startScopedSpan()) { tick0(); } } BackfillTriggerManager(StateManager stateManager,
                         Storage storage,
                         TriggerListener triggerListener,
                         Stats stats,
                         Time time); @VisibleForTesting  BackfillTriggerManager(StateManager stateManager,
                         Storage storage,
                         TriggerListener triggerListener,
                         Stats stats,
                         Time time,
                         Consumer<List<Backfill>> shuffler);   }
@Test public void shouldNotTriggerExecutionOnDisabledGlobally() throws IOException { when(config.globalEnabled()).thenReturn(false); backfillTriggerManager.tick(); verify(storage, never()).backfills(anyBoolean()); }
@Test public void shouldNotTriggerExecutionWhenFailedToReadConfig() throws IOException { when(storage.config()).thenThrow(new IOException()); backfillTriggerManager.tick(); verify(storage, never()).backfills(anyBoolean()); }
@Test public void shouldTriggerBackfillsNew() throws IOException { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); final int concurrency = BACKFILL_1.concurrency(); backfills.put(BACKFILL_1.id(), BACKFILL_1); backfillTriggerManager.tick(); final List<Instant> instants = instantsInRange(BACKFILL_1.start(), BACKFILL_1.end(), workflow.configuration().schedule()) .subList(0, concurrency); for (Instant instant : instants) { verify(triggerListener).event(workflow, Trigger.backfill(BACKFILL_1.id()), instant, TRIGGER_PARAMETERS); verify(transaction).store(BACKFILL_1.builder().nextTrigger(nextInstant(instant, BACKFILL_1.schedule())).build()); } verify(transaction, times(concurrency)).store(any(Backfill.class)); }
@Test public void shouldTriggerBackfillsNewReversed() throws IOException { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); final int concurrency = BACKFILL_2.concurrency(); backfills.put(BACKFILL_2.id(), BACKFILL_2); backfillTriggerManager.tick(); var schedule = workflow.configuration().schedule(); var instants = Lists.reverse(instantsInRange(BACKFILL_2.start(), BACKFILL_2.end(), schedule)) .subList(0, concurrency); for (final Instant instant : instants) { verify(triggerListener).event(workflow, Trigger.backfill(BACKFILL_2.id()), instant, TRIGGER_PARAMETERS); verify(transaction).store(BACKFILL_2.builder().nextTrigger(previousInstant(instant, BACKFILL_2.schedule())).build()); } verify(transaction, times(concurrency)).store(any(Backfill.class)); }
@Test public void shouldTriggerBackfillsInProgress() throws IOException { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); backfills.put(BACKFILL_1.id(), BACKFILL_1.builder() .nextTrigger(Instant.parse("2016-12-03T00:00:00Z")) .build()); final WorkflowInstance wfi1 = WorkflowInstance.create(WORKFLOW_ID1, "2016-12-02T23"); activeStates.put(wfi1, RunState.fresh(wfi1)); backfillTriggerManager.tick(); verify(triggerListener, only()).event( workflow, Trigger.backfill(BACKFILL_1.id()), Instant.parse("2016-12-03T00:00:00Z"), TRIGGER_PARAMETERS); }
@Test public void shouldTriggerBackfillsInProgressReversed() throws IOException { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); backfills.put(BACKFILL_2.id(), BACKFILL_2.builder() .nextTrigger(Instant.parse("2016-12-03T00:00:00Z")) .build()); final WorkflowInstance wfi1 = WorkflowInstance.create(WORKFLOW_ID1, "2016-12-03T01"); activeStates.put(wfi1, RunState.fresh(wfi1)); backfillTriggerManager.tick(); verify(triggerListener, only()).event( workflow, Trigger.backfill(BACKFILL_2.id()), Instant.parse("2016-12-03T00:00:00Z"), TRIGGER_PARAMETERS); }
@Test public void shouldTriggerBackfillsToCompletion() throws IOException { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); backfills.put(BACKFILL_3.id(), BACKFILL_3); backfillTriggerManager.tick(); var instants = instantsInRange(BACKFILL_3.start(), BACKFILL_3.end(), BACKFILL_3.schedule()); for (Instant instant : instants) { inOrder(triggerListener).verify(triggerListener) .event(workflow, Trigger.backfill(BACKFILL_3.id()), instant, TRIGGER_PARAMETERS); } var completedBackfill = BACKFILL_3.builder() .nextTrigger(BACKFILL_3.end()) .allTriggered(true) .build(); verify(transaction).store(completedBackfill); }
@Test public void shouldTriggerBackfillsToCompletionReversed() throws IOException { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); backfills.put(BACKFILL_4.id(), BACKFILL_4); backfillTriggerManager.tick(); var instants = Lists.reverse(instantsInRange(BACKFILL_4.start(), BACKFILL_4.end(), BACKFILL_4.schedule())); for (Instant instant : instants) { inOrder(triggerListener).verify(triggerListener) .event(workflow, Trigger.backfill(BACKFILL_4.id()), instant, TRIGGER_PARAMETERS); } var completedBackfill = BACKFILL_4.builder() .nextTrigger(previousInstant(BACKFILL_4.start(), BACKFILL_4.schedule())) .allTriggered(true) .build(); verify(transaction).store(completedBackfill); }
@Test public void shouldNotTriggerBackfillsIfNoCapacity() throws IOException { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); backfills.put(BACKFILL_1.id(), BACKFILL_1); final WorkflowInstance wfi1 = WorkflowInstance.create(WORKFLOW_ID1, "2016-12-02T22"); final WorkflowInstance wfi2 = WorkflowInstance.create(WORKFLOW_ID1, "2016-12-02T23"); activeStates.put(wfi1, RunState.fresh(wfi1)); activeStates.put(wfi2, RunState.fresh(wfi2)); backfillTriggerManager.tick(); verifyZeroInteractions(triggerListener); verify(transaction, never()).store(any(Backfill.class)); }
@Test public void shouldNotTriggerBackfillsIfNoCapacityReversed() throws IOException { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); backfills.put(BACKFILL_2.id(), BACKFILL_2); final WorkflowInstance wfi1 = WorkflowInstance.create(WORKFLOW_ID1, "2016-12-02T22"); final WorkflowInstance wfi2 = WorkflowInstance.create(WORKFLOW_ID1, "2016-12-02T23"); activeStates.put(wfi1, RunState.fresh(wfi1)); activeStates.put(wfi2, RunState.fresh(wfi2)); backfillTriggerManager.tick(); verifyZeroInteractions(triggerListener); verify(transaction, never()).store(any(Backfill.class)); }
@Test public void shouldNotTriggerIfAllTriggered() throws IOException { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); final Backfill completedBackfill = BACKFILL_3.builder().nextTrigger(BACKFILL_3.end()).allTriggered(true).build(); backfills.put(completedBackfill.id(), completedBackfill); backfillTriggerManager.tick(); verifyZeroInteractions(triggerListener); verify(transaction, never()).store(any(Backfill.class)); }
@Test public void shouldNotTriggerIfAllTriggeredReversed() throws IOException { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); final Backfill completedBackfill = BACKFILL_4.builder() .nextTrigger(previousInstant(BACKFILL_4.start(), BACKFILL_4.schedule())) .allTriggered(true) .build(); backfills.put(completedBackfill.id(), completedBackfill); backfillTriggerManager.tick(); verifyZeroInteractions(triggerListener); verify(transaction, never()).store(any(Backfill.class)); }
@Test public void shouldMarkAsAllTriggeredIfEndOfBackfillEncountered() throws IOException { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); final Backfill completedBackfill = BACKFILL_3.builder().nextTrigger(BACKFILL_3.end()).allTriggered(false).build(); backfills.put(completedBackfill.id(), completedBackfill); backfillTriggerManager.tick(); verify(transaction).store(completedBackfill.builder().allTriggered(true).build()); verifyZeroInteractions(triggerListener); }
@Test public void shouldMarkedAsAllTriggeredIfEndOfBackfillEncounteredReversed() throws IOException { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); final Backfill completedBackfill = BACKFILL_4.builder() .nextTrigger(previousInstant(BACKFILL_4.start(), BACKFILL_4.schedule())) .allTriggered(false) .build(); backfills.put(completedBackfill.id(), completedBackfill); backfillTriggerManager.tick(); verify(transaction).store(completedBackfill.builder().allTriggered(true).build()); verifyZeroInteractions(triggerListener); }
@Test public void shouldNotTriggerIfFailedToGetBackfills() throws Exception { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); doThrow(new IOException()).when(storage).backfills(anyBoolean()); backfillTriggerManager.tick(); verifyZeroInteractions(triggerListener); verify(transaction, never()).store(any(Backfill.class)); }
@Test public void shouldNotTriggerBackfillsWithMissingWorkflows() throws Exception { backfills.put(BACKFILL_1.id(), BACKFILL_1); when(transaction.workflow(BACKFILL_1.workflowId())).thenReturn(Optional.empty()); backfillTriggerManager.tick(); verifyZeroInteractions(triggerListener); verify(transaction).store(BACKFILL_1.builder().halted(true).build()); verify(transaction, times(1)).store(any(Backfill.class)); }
@Test public void shouldNotTriggerBackfillsWithMissingBackfill() throws Exception { backfills.put(BACKFILL_1.id(), BACKFILL_1); when(transaction.backfill(BACKFILL_1.id())).thenReturn(Optional.empty()); backfillTriggerManager.tick(); verifyZeroInteractions(triggerListener); }
@Test public void shouldNotTriggerBackfillsAndStoreBackfillWithMissingWorkflows() throws Exception { backfills.put(BACKFILL_1.id(), BACKFILL_1); when(transaction.workflow(BACKFILL_1.workflowId())).thenReturn(Optional.empty()); backfillTriggerManager.tick(); verifyZeroInteractions(triggerListener); verify(transaction).store(BACKFILL_1.builder().halted(true).build()); verify(transaction, times(1)).store(any(Backfill.class)); }
@Test public void shouldNotTriggerBackfillsWhenFailedToReadWorkflow() throws Exception { backfills.put(BACKFILL_1.id(), BACKFILL_1); when(transaction.workflow(BACKFILL_1.workflowId())).thenThrow(new IOException()); backfillTriggerManager.tick(); verifyZeroInteractions(triggerListener); verify(transaction, never()).store(any(Backfill.class)); }
@Test public void shouldNotTriggerBackfillsIfRunInTransactionFails() throws Exception { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); doThrow(new IOException()).when(storage).runInTransactionWithRetries(any()); backfills.put(BACKFILL_1.id(), BACKFILL_1); backfillTriggerManager.tick(); verifyZeroInteractions(triggerListener); }
@Test public void shouldContinueTriggeringNextBackfillWhenUnknownExecutionException() throws IOException { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); final int concurrency = BACKFILL_1.concurrency(); backfills.put(BACKFILL_1.id(), BACKFILL_1); backfills.put(BACKFILL_3.id(), BACKFILL_3); doThrow(new RuntimeException()) .when(triggerListener) .event(any(), any(), eq(Instant.parse("2016-12-02T22:00:00Z")), eq(TRIGGER_PARAMETERS)); backfillTriggerManager.tick(); final List<Instant> instants = instantsInRange(BACKFILL_3.start(), BACKFILL_3.end(), workflow.configuration().schedule()); instants.stream().limit(concurrency).forEach(instant -> verify(triggerListener).event(workflow, Trigger.backfill(BACKFILL_3.id()), instant, TRIGGER_PARAMETERS)); final Backfill completedBackfill = BACKFILL_3.builder().nextTrigger(BACKFILL_3.end()).allTriggered(true).build(); verify(transaction).store(completedBackfill); }
@Test public void shouldContinueTriggeringNextBackfillWhenOtherException() throws IOException { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); final int concurrency = BACKFILL_1.concurrency(); backfills.put(BACKFILL_1.id(), BACKFILL_1); backfills.put(BACKFILL_3.id(), BACKFILL_3); doThrow(new RuntimeException()) .when(triggerListener) .event(any(), any(), eq(Instant.parse("2016-12-02T22:00:00Z")), eq(TRIGGER_PARAMETERS)); backfillTriggerManager.tick(); final List<Instant> instants = instantsInRange(BACKFILL_3.start(), BACKFILL_3.end(), workflow.configuration().schedule()); instants.stream().limit(concurrency).forEach(instant -> verify(triggerListener).event(workflow, Trigger.backfill( BACKFILL_3.id()), instant, TRIGGER_PARAMETERS)); final Backfill completedBackfill = BACKFILL_3.builder().nextTrigger(BACKFILL_3.end()).allTriggered(true).build(); verify(transaction).store(completedBackfill); }
@Test public void shouldIgnoreIfAlreadyInitialized() throws IOException { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); final int concurrency = BACKFILL_1.concurrency(); backfills.put(BACKFILL_1.id(), BACKFILL_1); doThrow(new AlreadyInitializedException("")) .when(triggerListener) .event(any(), any(), eq(Instant.parse("2016-12-02T22:00:00Z")), eq(TRIGGER_PARAMETERS)); backfillTriggerManager.tick(); final List<Instant> instants = instantsInRange(BACKFILL_1.start(), BACKFILL_1.end(), workflow.configuration().schedule()); instants.stream().limit(concurrency).forEach(instant -> verify(triggerListener).event(workflow, Trigger.backfill(BACKFILL_1.id()), instant, TRIGGER_PARAMETERS)); verify(transaction) .store(BACKFILL_1.builder().nextTrigger(instants.get(concurrency)).build()); }