BackfillTriggerManager { @VisibleForTesting boolean triggerNextInstanceAndProgress(StorageTransaction tx, String backfillId, Workflow workflow, int triggeredInstances) throws IOException { var backfillOpt = tx.backfill(backfillId); if (backfillOpt.isEmpty()) { return false; } var backfill = backfillOpt.orElseThrow(); if (backfill.halted()) { LOG.debug("Backfill halted: {}", backfill); return false; } if (triggeredInstances >= backfill.concurrency()) { LOG.debug("Capacity reached for backfill: {}", backfill); return false; } final Instant nextTrigger = backfill.nextTrigger(); if (isAllTriggered(backfill, nextTrigger)) { LOG.debug("Backfill already completely triggered: {}", backfill); if (!backfill.allTriggered()) { tx.store(backfill.builder().allTriggered(true).build()); } return false; } try { final TriggerParameters parameters = backfill.triggerParameters().orElse(TriggerParameters.zero()); triggerListener.event(workflow, Trigger.backfill(backfill.id()), nextTrigger, parameters); } catch (Exception e) { if (findCause(e, AlreadyInitializedException.class) != null) { LOG.debug("{} already triggered for backfill {}", nextTrigger, backfill, e); } else { LOG.debug("Failed to trigger {} for backfill {}", nextTrigger, backfill, e); throw new RuntimeException(e); } } var nextTriggerInstant = nextTriggerInstant(backfill, nextTrigger); var nextBackfill = backfill.builder() .nextTrigger(nextTriggerInstant) .allTriggered(isAllTriggered(backfill, nextTriggerInstant)) .build(); tx.store(nextBackfill); if (nextBackfill.allTriggered()) { LOG.debug("Backfill all triggered: {}", backfill.id()); } return !nextBackfill.allTriggered(); } 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 shouldNotTriggerNextPartitionAndProgressIfBackfillHalted() throws Exception { final Workflow workflow = createWorkflow(WORKFLOW_ID1); initWorkflow(workflow); backfills.put(BACKFILL_1.id(), BACKFILL_1.builder().halted(true).build()); var moveOn = backfillTriggerManager.triggerNextInstanceAndProgress(transaction, BACKFILL_1.id(), workflow, 0); assertFalse(moveOn); verifyNoMoreInteractions(triggerListener); }