LanderTaskRunner { public boolean run(OffsetDateTime runtimeDateTime) { emitter .emit(new PatchSet(roadName, singletonList(PatchOperation.replace(LAST_RUN_PATH, runtimeDateTime.toString())))); String acquisitionInstant = ISODateTimeFormat.basicDateTimeNoMillis().withZoneUTC().print( runtimeDateTime.toInstant().toEpochMilli()); changeState(State.PREPARING); log.info("Preparing to land partition {}.", acquisitionInstant); boolean runAgain = landingTimer.record(() -> runChain(acquisitionInstant)); log.info("Update message for road {}: landerLastRun: {}, runAgain: {}", roadName, runtimeDateTime, runAgain); return runAgain; } LanderTaskRunner(
      MeterRegistry registry,
      OffsetManager offsetManager,
      String roadName,
      String topicName,
      String database,
      HivePartitionManager hivePartitionManager,
      Lander.Factory landerFactory,
      HiveNotificationHandler landingHandler,
      PatchSetEmitter emitter,
      Clock clock,
      long maxRecordsPerPartition,
      boolean enableServerSideEncryption,
      int landingTimeoutMinutes); boolean run(OffsetDateTime runtimeDateTime); static final String ACQUISITION_INSTANT; static final String REVISION_INSTANT; }
@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); }