AlertsUIServiceImpl implements AlertsUIService { @Override public void escalateAlerts(List<Map<String, Object>> alerts) throws RestException { String user = SecurityUtils.getCurrentUser(); String topic = environment.getProperty(KAFKA_TOPICS_ESCALATION_PROPERTY); Long now = clock.currentTimeMillis(); LOG.info("Escalating {} alert(s): user={}, topic={}, timestamp={}", alerts.size(), user, topic, now); try { for (Map<String, Object> alert : alerts) { alert.put(METRON_ESCALATION_USER_FIELD, user); alert.put(METRON_ESCALATION_TIMESTAMP_FIELD, now); String message = JSONUtils.INSTANCE.toJSON(alert, false); kafkaService.produceMessage(topic, message); } } catch (JsonProcessingException e) { throw new RestException(e); } } @Autowired  AlertsUIServiceImpl(final KafkaService kafkaService,
                             final Environment environment,
                             final UserSettingsClient userSettingsClient); @Override void escalateAlerts(List<Map<String, Object>> alerts); @Override Optional<AlertsUIUserSettings> getAlertsUIUserSettings(); @Override Map<String, AlertsUIUserSettings> findAllAlertsUIUserSettings(); @Override void saveAlertsUIUserSettings(AlertsUIUserSettings alertsUIUserSettings); @Override boolean deleteAlertsUIUserSettings(String user); void setClock(Clock clock); static final String ALERT_USER_SETTING_TYPE; static ThreadLocal<ObjectMapper> _mapper; }
@Test public void escalateAlertShouldSendMessageToKafka() throws Exception { final String field = "field"; final String value1 = "value1"; final String value2 = "value2"; final String escalationTopic = "escalation"; when(environment.getProperty(MetronRestConstants.KAFKA_TOPICS_ESCALATION_PROPERTY)).thenReturn(escalationTopic); final Map<String, Object> alert1 = mapOf(field, value1); String escalationMessage1 = escalationMessage(field, value1, user1, clock.currentTimeMillis()); final Map<String, Object> alert2 = mapOf(field, value2); String escalationMessage2 = escalationMessage(field, value2, user1, clock.currentTimeMillis()); alertsUIService.escalateAlerts(Arrays.asList(alert1, alert2)); verify(kafkaService).produceMessage(escalationTopic, escalationMessage1); verify(kafkaService).produceMessage(escalationTopic, escalationMessage2); verifyNoMoreInteractions(kafkaService); }