LanderTaskRunner { LanderConfiguration prepareLanderConfiguration(String acquisitionInstant) { Map<Integer, Long> comittedOffsets = offsetManager.getCommittedOffsets(topicName); Map<Integer, OffsetRange> offsets = new HashMap<>(); boolean runAgain = false; for (Entry<Integer, Long> entry : offsetManager.getLatestOffsets(topicName).entrySet()) { Integer partition = entry.getKey(); long committedOffset = comittedOffsets.getOrDefault(partition, 0L); long latestOffset = entry.getValue(); if (latestOffset > committedOffset) { if (latestOffset > committedOffset + maxRecordsPerPartition) { latestOffset = committedOffset + maxRecordsPerPartition; runAgain = true; } offsets.put(partition, new OffsetRange(committedOffset, latestOffset)); } } if (offsets.isEmpty()) { throw new NoDataToLandException(); } String s3KeyPrefix = String.format("%s/%s/%d/%s=%s", database, roadName, clock.millis(), ACQUISITION_INSTANT, acquisitionInstant); changeState(State.LANDING); log.info("Landing partition {}.", acquisitionInstant); return new LanderConfiguration(roadName, topicName, offsets, s3KeyPrefix, enableServerSideEncryption, acquisitionInstant, 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 prepareLanderConfiguration() { Map<Integer, Long> latestOffsets = ImmutableMap.of(1, 10L, 2, 20L); Map<Integer, Long> committedOffsets = ImmutableMap.of(1, 5L, 2, 15L); when(offsetManager.getLatestOffsets(TOPIC_NAME)).thenReturn(latestOffsets); when(offsetManager.getCommittedOffsets(TOPIC_NAME)).thenReturn(committedOffsets); assertThat(underTest.getState(), is(State.IDLE)); LanderConfiguration landerConfiguration = underTest.prepareLanderConfiguration(ACQUISITION_INSTANT); assertThat(landerConfiguration, is(expectedLanderConfiguration)); assertThat(underTest.getState(), is(State.LANDING)); }