LoadingBay implements Agent<HiveRoad> { @Override public List<PatchOperation> inspectModel(String key, HiveRoad model) { Optional<Hive> hive = Optional.ofNullable(model.getDestinations()).map(Destinations::getHive); if (hive.isPresent()) { OffsetDateTime lastRun = hive.map(Hive::getStatus).map(HiveStatus::getLastRun).orElse(EPOCH); log.debug("Inspecting road {}. Lander last ran at {}", model.getName(), lastRun); try { List<PatchOperation> patches = hiveTableAction.checkAndApply(model); LanderMonitor monitor = monitors.computeIfAbsent(model.getName(), n -> monitorFactory.apply(model)); monitor.establishLandingFrequency(hive.map(Hive::getLandingInterval).orElse(Hive.DEFAULT_LANDING_INTERVAL)); monitor.setEnabled(hive.get().isEnabled()); return patches; } catch (NoActiveSchemaException e) { log.info("No schema defined on road '{}'", model.getName()); return emptyList(); } catch (Exception e) { log.error("Error while applying actions for road '{}'", model.getName(), e); return emptyList(); } } else { log.info("Skipping road {} because it has no Hive destination", model.getName()); if (monitors.containsKey(model.getName())) { try { monitors.remove(model.getName()).close(); } catch (Exception e) { log.warn("Error shutting down DestinationMonitor for {}", model.getName(), e); } } return emptyList(); } } @Autowired  LoadingBay(HiveTableAction hiveTableAction, Function<HiveRoad, LanderMonitor> monitorFactory); @Override List<PatchOperation> newModel(String key, HiveRoad newModel); @Override List<PatchOperation> updatedModel(String key, HiveRoad oldModel, HiveRoad newModel); @Override void deletedModel(String key, HiveRoad oldModel); @Override List<PatchOperation> inspectModel(String key, HiveRoad model);  }
@Test public void noHiveDestination() { Destinations destinations = Destinations.builder().build(); HiveRoad road = HiveRoad.builder().name(ROAD_NAME).destinations(destinations).build(); List<PatchOperation> operations = underTest.inspectModel(ROAD_NAME, road); assertThat(operations.size(), is(0)); }
@Test public void actionsCreated() { Hive hive = Hive.builder().status(HiveStatus.builder().build()).build(); Destinations destinations = Destinations.builder().hive(hive).build(); HiveRoad road = HiveRoad.builder().name(ROAD_NAME).destinations(destinations).build(); when(monitorFactory.apply(road)).thenReturn(landerMonitor); List<PatchOperation> operations = underTest.inspectModel(ROAD_NAME, road); assertThat(operations.size(), is(0)); verify(landerMonitor).setEnabled(false); verify(landerMonitor).establishLandingFrequency("PT1H"); }
@Test public void enableHiveDestination() { Hive hive = Hive.builder().enabled(true).status(HiveStatus.builder().build()).build(); Destinations destinations = Destinations.builder().hive(hive).build(); HiveRoad road = HiveRoad.builder().name(ROAD_NAME).destinations(destinations).build(); when(monitorFactory.apply(road)).thenReturn(landerMonitor); List<PatchOperation> operations = underTest.inspectModel(ROAD_NAME, road); assertThat(operations.size(), is(0)); verify(landerMonitor).setEnabled(true); }
@Test public void customLandingInterval() { Hive hive = Hive.builder().landingInterval(CUSTOM_LANDING_INTERVAL).status(HiveStatus.builder().build()).build(); Destinations destinations = Destinations.builder().hive(hive).build(); HiveRoad road = HiveRoad.builder().name(ROAD_NAME).destinations(destinations).build(); when(monitorFactory.apply(road)).thenReturn(landerMonitor); List<PatchOperation> operations = underTest.inspectModel(ROAD_NAME, road); assertThat(operations.size(), is(0)); verify(landerMonitor).establishLandingFrequency(CUSTOM_LANDING_INTERVAL); }
@Test public void landerLastRunPresent() { Hive hive = Hive .builder() .landingInterval(CUSTOM_LANDING_INTERVAL) .status( HiveStatus.builder().lastRun(OffsetDateTime.ofInstant(Instant.ofEpochMilli(123L), ZoneOffset.UTC)).build()) .build(); Destinations destinations = Destinations.builder().hive(hive).build(); HiveRoad road = HiveRoad.builder().name(ROAD_NAME).destinations(destinations).build(); when(monitorFactory.apply(road)).thenReturn(landerMonitor); List<PatchOperation> operations = underTest.inspectModel(ROAD_NAME, road); assertThat(operations.size(), is(0)); verify(landerMonitor).establishLandingFrequency(CUSTOM_LANDING_INTERVAL); }
@Test public void exceptionDuringAction() { RuntimeException e = new RuntimeException("foo"); doThrow(e).when(hiveTableAction).checkAndApply(any(HiveRoad.class)); Hive hive = Hive.builder().status(HiveStatus.builder().build()).build(); Destinations destinations = Destinations.builder().hive(hive).build(); HiveRoad road = HiveRoad.builder().name(ROAD_NAME).destinations(destinations).build(); List<PatchOperation> operations = underTest.inspectModel(ROAD_NAME, road); assertThat(operations.size(), is(0)); }