KafkaEmitter implements ProfileMeasurementEmitter, Serializable { @Override public void emit(ProfileMeasurement measurement, OutputCollector collector) { Map<String, Object> triageValues = measurement.getTriageValues(); if(MapUtils.isNotEmpty(triageValues)) { JSONObject message = createMessage(measurement); appendTriageValues(measurement, message); collector.emit(getStreamId(), new Values(message)); LOG.debug("Emitted measurement; stream={}, profile={}, entity={}, period={}, start={}, end={}", getStreamId(), measurement.getProfileName(), measurement.getEntity(), measurement.getPeriod().getPeriod(), measurement.getPeriod().getStartTimeMillis(), measurement.getPeriod().getEndTimeMillis()); } else { LOG.debug("No triage values, nothing to emit; stream={}, profile={}, entity={}, period={}, start={}, end={}", getStreamId(), measurement.getProfileName(), measurement.getEntity(), measurement.getPeriod().getPeriod(), measurement.getPeriod().getStartTimeMillis(), measurement.getPeriod().getEndTimeMillis()); } }  @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override void emit(ProfileMeasurement measurement, OutputCollector collector); @Override String getStreamId(); void setStreamId(String streamId); String getSourceType(); void setSourceType(String sourceType); static final String PROFILE_FIELD; static final String ENTITY_FIELD; static final String PERIOD_ID_FIELD; static final String PERIOD_START_FIELD; static final String PERIOD_END_FIELD; static final String TIMESTAMP_FIELD; static final String ALERT_FIELD; }
@Test public void testEmit() { ProfileMeasurement measurement = new ProfileMeasurement() .withProfileName("profile") .withEntity("entity") .withPeriod(20000, 15, TimeUnit.MINUTES) .withDefinition(profile) .withTriageValues(Collections.singletonMap("triage-key", "triage-value")); kafkaEmitter.emit(measurement, collector); verify(collector, times(1)).emit(eq(kafkaEmitter.getStreamId()), any()); }
@Test public void testTriageValueInMessage() { ProfileMeasurement measurement = new ProfileMeasurement() .withDefinition(profile) .withProfileName(profile.getProfile()) .withEntity("entity") .withPeriod(20000, 15, TimeUnit.MINUTES) .withTriageValues(Collections.singletonMap("triage-key", "triage-value")); kafkaEmitter.emit(measurement, collector); JSONObject actual = expectJsonObject(kafkaEmitter, collector); assertEquals(measurement.getProfileName(), actual.get("profile")); assertEquals(measurement.getEntity(), actual.get("entity")); assertEquals(measurement.getPeriod().getPeriod(), actual.get("period")); assertEquals(measurement.getPeriod().getStartTimeMillis(), actual.get("period.start")); assertEquals(measurement.getPeriod().getEndTimeMillis(), actual.get("period.end")); assertEquals("profiler", actual.get("source.type")); assertNotNull(actual.get("timestamp")); assertNotNull(actual.get(Constants.GUID)); assertEquals(measurement.getTriageValues().get("triage-key"), actual.get("triage-key")); }
@Test public void testMultipleTriageValueInMessage() { Map<String, Object> triageValues = ImmutableMap.of( "x", 2, "y", "4", "z", 6.0); ProfileMeasurement measurement = new ProfileMeasurement() .withDefinition(profile) .withProfileName(profile.getProfile()) .withEntity("entity") .withPeriod(20000, 15, TimeUnit.MINUTES) .withTriageValues(triageValues); kafkaEmitter.emit(measurement, collector); JSONObject actual = expectJsonObject(kafkaEmitter, collector); assertEquals(measurement.getTriageValues().get("x"), actual.get("x")); assertEquals(measurement.getTriageValues().get("y"), actual.get("y")); assertEquals(measurement.getTriageValues().get("z"), actual.get("z")); }
@Test public void testInvalidType() { Map<String, Object> triageValues = ImmutableMap.of( "invalid", new OnlineStatisticsProvider(), "valid", 4); ProfileMeasurement measurement = new ProfileMeasurement() .withDefinition(profile) .withProfileName(profile.getProfile()) .withEntity("entity") .withPeriod(20000, 15, TimeUnit.MINUTES) .withTriageValues(triageValues); kafkaEmitter.emit(measurement, collector); JSONObject actual = expectJsonObject(kafkaEmitter, collector); assertEquals(measurement.getProfileName(), actual.get("profile")); assertEquals(measurement.getEntity(), actual.get("entity")); assertEquals(measurement.getPeriod().getPeriod(), actual.get("period")); assertEquals(measurement.getPeriod().getStartTimeMillis(), actual.get("period.start")); assertEquals(measurement.getPeriod().getEndTimeMillis(), actual.get("period.end")); assertEquals("profiler", actual.get("source.type")); assertNotNull(actual.get("timestamp")); assertNotNull(actual.get(Constants.GUID)); assertFalse(actual.containsKey("invalid")); assertEquals(triageValues.get("valid"), actual.get("valid")); }
@Test public void testIntegerIsValidType() { ProfileMeasurement measurement = new ProfileMeasurement() .withDefinition(profile) .withProfileName(profile.getProfile()) .withEntity("entity") .withPeriod(20000, 15, TimeUnit.MINUTES) .withTriageValues(Collections.singletonMap("triage-key", 123)); kafkaEmitter.emit(measurement, collector); JSONObject actual = expectJsonObject(kafkaEmitter, collector); assertEquals(measurement.getTriageValues().get("triage-key"), actual.get("triage-key")); }
@Test public void testStringIsValidType() { ProfileMeasurement measurement = new ProfileMeasurement() .withDefinition(profile) .withProfileName(profile.getProfile()) .withEntity("entity") .withPeriod(20000, 15, TimeUnit.MINUTES) .withTriageValues(Collections.singletonMap("triage-key", "value")); kafkaEmitter.emit(measurement, collector); JSONObject actual = expectJsonObject(kafkaEmitter, collector); assertEquals(measurement.getTriageValues().get("triage-key"), actual.get("triage-key")); }