@Test
public void runTypical() throws Exception {

  when(offsetManager.getLatestOffsets(TOPIC_NAME)).thenReturn(ImmutableMap.of(1, 10L, 2, 20L));
  when(offsetManager.getCommittedOffsets(TOPIC_NAME)).thenReturn(ImmutableMap.of(1, 5L, 2, 15L));
  when(landerFactory.newInstance(expectedLanderConfiguration)).thenReturn(lander);
  CompletableFuture<LanderConfiguration> future = new CompletableFuture<>();
  when(lander.run()).thenReturn(future);
  future.complete(expectedLanderConfiguration);
  when(hivePartitionManager.addPartition(eq(ROAD_NAME), any(), any())).thenReturn(Optional.of(partition));
  underTest.run(OffsetDateTime.ofInstant(Instant.ofEpochMilli(runtimeMillis), ZoneOffset.UTC));
  verify(offsetManager).getCommittedOffsets(TOPIC_NAME);
  verify(offsetManager).getLatestOffsets(TOPIC_NAME);
  verify(landerFactory).newInstance(expectedLanderConfiguration);
  verify(hivePartitionManager).addPartition(ROAD_NAME, singletonList(ACQUISITION_INSTANT), S3_PREFIX);
  verify(offsetManager).commitOffsets(TOPIC_NAME, ImmutableMap.of(1, 10L, 2, 20L));
  verify(landingNotifier).handlePartitionCreated(ROAD_NAME, partition, PARTITION_SPEC, 10L);
  verify(emitter)
      .emit(new PatchSet(ROAD_NAME, singletonList(PatchOperation.replace(LAST_RUN_PATH, "2018-05-16T09:17:04Z"))));
  verifyNoMoreInteractions(emitter);
}