ReplayEvents { public static Optional<RunStateData> getBackfillRunStateData( WorkflowInstance workflowInstance, Storage storage, String backfillId) { final SortedSet<SequenceEvent> sequenceEvents = getSequenceEvents(workflowInstance, storage); if (sequenceEvents.isEmpty()) { return Optional.empty(); } final SettableTime time = new SettableTime(); RunState restoredState = RunState.fresh(workflowInstance, time); boolean backfillFound = false; final SettableTime initialTime = new SettableTime(); final SettableTime latestTime = new SettableTime(); for (SequenceEvent sequenceEvent : sequenceEvents) { time.set(Instant.ofEpochMilli(sequenceEvent.timestamp())); final boolean triggerExecutionEventMet = "triggerExecution".equals(EventUtil.name(sequenceEvent.event())); if (triggerExecutionEventMet) { if (backfillFound) { break; } restoredState = RunState.fresh(workflowInstance, time); initialTime.set(time.get()); } try { restoredState = restoredState.transition(sequenceEvent.event(), time); } catch (IllegalStateException e) { LOG.warn("failed to transition state, move on to next event", e); } latestTime.set(time.get()); if (backfillFound(triggerExecutionEventMet, backfillId, restoredState)) { backfillFound = true; } } if (backfillFound) { final RunStateData runStateData = RunStateData.newBuilder() .workflowInstance(restoredState.workflowInstance()) .state(restoredState.state().name()) .stateData(restoredState.data()) .initialTimestamp(initialTime.get().toEpochMilli()) .latestTimestamp(latestTime.get().toEpochMilli()) .build(); return Optional.of(runStateData); } else { return Optional.empty(); } } private  ReplayEvents(); static Optional<RunStateData> getBackfillRunStateData(
      WorkflowInstance workflowInstance,
      Storage storage,
      String backfillId);  }
@Test public void restoreRunStateForInactiveBackfill() throws Exception { SortedSet<SequenceEvent> events = newTreeSet(SequenceEvent.COUNTER_COMPARATOR); events.add(SequenceEvent.create( Event.triggerExecution(WORKFLOW_INSTANCE, Trigger.backfill("bf-1"), TRIGGER_PARAMETERS), 1L, 1L)); events.add(SequenceEvent.create(Event.dequeue(WORKFLOW_INSTANCE, RESOURCE_IDS), 2L, 2L)); events.add(SequenceEvent.create(Event.submit(WORKFLOW_INSTANCE, EXECUTION_DESCRIPTION, "exec-1"), 3L, 3L)); events.add(SequenceEvent.create(Event.submitted(WORKFLOW_INSTANCE, "exec-1", "test"), 4L, 4L)); events.add(SequenceEvent.create(Event.started(WORKFLOW_INSTANCE), 5L, 5L)); events.add(SequenceEvent.create(Event.terminate(WORKFLOW_INSTANCE, Optional.of(0)), 6L, 6L)); events.add(SequenceEvent.create(Event.success(WORKFLOW_INSTANCE), 7L, 7L)); events.add(SequenceEvent.create( Event.triggerExecution(WORKFLOW_INSTANCE, Trigger.adhoc("ad-hoc"), TRIGGER_PARAMETERS), 8L, 8L)); events.add(SequenceEvent.create(Event.dequeue(WORKFLOW_INSTANCE, RESOURCE_IDS), 9L, 9L)); events.add(SequenceEvent.create(Event.halt(WORKFLOW_INSTANCE), 10L, 10L)); when(storage.readEvents(WORKFLOW_INSTANCE)).thenReturn(events); RunStateData restoredRunStateData = ReplayEvents.getBackfillRunStateData(WORKFLOW_INSTANCE, storage, "bf-1").orElseThrow(); assertThat(restoredRunStateData.state(), is("DONE")); assertThat(restoredRunStateData.stateData().lastExit(), isPresent()); assertThat(restoredRunStateData.stateData().lastExit().orElseThrow(), is(0)); assertThat(restoredRunStateData.initialTimestamp().orElseThrow(), is(1L)); assertThat(restoredRunStateData.latestTimestamp().orElseThrow(), is(7L)); }
@Test public void restoreRunStateWhenMissingEvent() throws Exception { SortedSet<SequenceEvent> events = newTreeSet(SequenceEvent.COUNTER_COMPARATOR); events.add(SequenceEvent.create( Event.triggerExecution(WORKFLOW_INSTANCE, Trigger.backfill("bf-1"), TRIGGER_PARAMETERS), 1L, 1L)); events.add(SequenceEvent.create(Event.dequeue(WORKFLOW_INSTANCE, RESOURCE_IDS), 2L, 2L)); events.add(SequenceEvent.create(Event.submitted(WORKFLOW_INSTANCE, "exec-1", "test"), 4L, 4L)); events.add(SequenceEvent.create(Event.started(WORKFLOW_INSTANCE), 5L, 5L)); events.add(SequenceEvent.create(Event.terminate(WORKFLOW_INSTANCE, Optional.of(0)), 6L, 6L)); events.add(SequenceEvent.create(Event.success(WORKFLOW_INSTANCE), 7L, 7L)); events.add(SequenceEvent.create( Event.triggerExecution(WORKFLOW_INSTANCE, Trigger.adhoc("ad-hoc"), TRIGGER_PARAMETERS), 8L, 8L)); events.add(SequenceEvent.create(Event.dequeue(WORKFLOW_INSTANCE, RESOURCE_IDS), 9L, 9L)); events.add(SequenceEvent.create(Event.halt(WORKFLOW_INSTANCE), 10L, 10L)); when(storage.readEvents(WORKFLOW_INSTANCE)).thenReturn(events); RunStateData restoredRunStateData = ReplayEvents.getBackfillRunStateData(WORKFLOW_INSTANCE, storage, "bf-1").orElseThrow(); assertThat(restoredRunStateData.state(), is("DONE")); assertThat(restoredRunStateData.stateData().lastExit(), isPresent()); assertThat(restoredRunStateData.stateData().lastExit().orElseThrow(), is(0)); assertThat(restoredRunStateData.initialTimestamp().orElseThrow(), is(1L)); assertThat(restoredRunStateData.latestTimestamp().orElseThrow(), is(7L)); }
@Test public void returnsEmptyWithMissingBackfill() throws Exception { SortedSet<SequenceEvent> events = newTreeSet(SequenceEvent.COUNTER_COMPARATOR); events.add(SequenceEvent.create( Event.triggerExecution(WORKFLOW_INSTANCE, Trigger.backfill("bf-1"), TRIGGER_PARAMETERS), 1L, 1L)); events.add(SequenceEvent.create(Event.dequeue(WORKFLOW_INSTANCE, RESOURCE_IDS), 2L, 2L)); when(storage.readEvents(WORKFLOW_INSTANCE)).thenReturn(events); Optional<RunStateData> restoredRunStateData = ReplayEvents.getBackfillRunStateData( WORKFLOW_INSTANCE, storage, "erroneous-id"); assertThat(restoredRunStateData, is(Optional.empty())); }
@Test public void returnsEmptyIfNoEvent() throws Exception { SortedSet<SequenceEvent> events = newTreeSet(SequenceEvent.COUNTER_COMPARATOR); when(storage.readEvents(WORKFLOW_INSTANCE)).thenReturn(events); Optional<RunStateData> restoredRunStateData = ReplayEvents.getBackfillRunStateData( WORKFLOW_INSTANCE, storage, "bf-1"); assertThat(restoredRunStateData, is(Optional.empty())); }
@Test public void throwExceptionWhenFailedToReadEvents() throws Exception { IOException exception = new IOException(); when(storage.readEvents(WORKFLOW_INSTANCE)).thenThrow(exception); try { ReplayEvents.getBackfillRunStateData(WORKFLOW_INSTANCE, storage, "bf-1"); fail(); } catch (RuntimeException e) { assertThat(e.getCause(), is(exception)); } }