@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)); }
@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(); } }
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(); } } }
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); }
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); }
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);  }