LanderTaskRunner { boolean runChain(String acquisitionInstant) { try { LanderConfiguration landerConfiguration = prepareLanderConfiguration(acquisitionInstant); CompletableFuture<LanderConfiguration> future = landerFactory.newInstance(landerConfiguration).run(); try { future.get(landingTimeoutMinutes, MINUTES); updateMetadata(landerConfiguration); long totalMessages = landerConfiguration .getOffsets() .values() .stream() .mapToLong(r -> r.getEnd() - r.getStart()) .sum(); messagesLandedCounter.increment(totalMessages); } catch (TimeoutException e) { log.warn("Landing of {}, {} timed out", landerConfiguration.getRoadName(), acquisitionInstant); future.cancel(true); return true; } catch (MetaStoreException e) { return true; } finally { changeState(State.IDLE); } return landerConfiguration.isRunAgain(); } catch (Throwable t) { onException(acquisitionInstant, t); } return false; } 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 runChainTypical() throws Exception { underTest.runChain(partitionColumnValue); }
@Test public void runChainNoDataToLandException() throws Exception { when(offsetManager.getCommittedOffsets(TOPIC_NAME)).thenReturn(singletonMap(0, 100L)); when(offsetManager.getLatestOffsets(TOPIC_NAME)).thenReturn(singletonMap(0, 100L)); underTest.runChain(partitionColumnValue); verify(emitter, never()).emit(any()); }