StreamlineEventLogger extends FileBasedEventLogger { @Override protected String buildLogMessage(EventInfo event) { String timestampStr = dateFormat.format(event.getTs()); List<Object> values = event.getValues(); if (!values.isEmpty()) { final Object eventObj = values.get(0); if (eventObj instanceof StreamlineEvent) { final StreamlineEvent slEvent = (StreamlineEvent) eventObj; Set<String> rootIds; if (EventCorrelationInjector.containsRootIds(slEvent)) { rootIds = EventCorrelationInjector.getRootIds(slEvent); } else { rootIds = Collections.emptySet(); } Set<String> parentIds; if (EventCorrelationInjector.containsParentIds(slEvent)) { parentIds = EventCorrelationInjector.getParentIds(slEvent); } else { parentIds = Collections.emptySet(); } String format = String.join(DELIMITER, new String[] { "%s","%s","%s","%s","%s","%s","%s","%s","%s"} ); return String.format(format, timestampStr, MARKER_FOR_STREAMLINE_EVENT, StormTopologyUtil.extractStreamlineComponentName(event.getComponent()), slEvent.getId(), rootIds, parentIds, ImmutableMap.copyOf(slEvent), slEvent.getHeader().toString(), slEvent.getAuxiliaryFieldsAndValues().toString()); } } Object messageId = event.getMessageId(); return String.format("%s,%s,%s,%s,%s,%s", timestampStr, MARKER_FOR_OTHER_EVENT, event.getComponent(), String.valueOf(event.getTask()), (messageId == null ? "" : messageId.toString()), values); }   static final String MARKER_FOR_STREAMLINE_EVENT; static final String MARKER_FOR_OTHER_EVENT; static final String DELIMITER; }
@Test public void buildStreamlineEventLogMessage() { long timestamp = 1513300663123L; String timestampStr = dateFormat.format(timestamp); String stormComponentName = "1-Component"; String streamlineComponent = "Component"; int taskId = 2; Object messageId = "dummy"; StreamlineEvent event = buildTestEvent(); List<Object> values = Collections.singletonList(event); String expectMessage = String.format("%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s", timestampStr, StreamlineEventLogger.DELIMITER, StreamlineEventLogger.MARKER_FOR_STREAMLINE_EVENT, StreamlineEventLogger.DELIMITER, streamlineComponent, StreamlineEventLogger.DELIMITER, event.getId(), StreamlineEventLogger.DELIMITER, "[]", StreamlineEventLogger.DELIMITER, "[]", StreamlineEventLogger.DELIMITER, ImmutableMap.copyOf(event), StreamlineEventLogger.DELIMITER, event.getHeader(), StreamlineEventLogger.DELIMITER, event.getAuxiliaryFieldsAndValues()); IEventLogger.EventInfo eventInfo = new IEventLogger.EventInfo(timestamp, stormComponentName, taskId, messageId, values); String actualMessage = sut.buildLogMessage(eventInfo); Assert.assertEquals(expectMessage, actualMessage); EventCorrelationInjector eventCorrelationInjector = new EventCorrelationInjector(); StreamlineEvent event2 = StreamlineEventImpl.builder().from(event).build(); StreamlineEvent event3 = StreamlineEventImpl.builder().from(event).build(); event2 = eventCorrelationInjector.injectCorrelationInformation(event2, Collections.singletonList(event3), streamlineComponent); event = eventCorrelationInjector.injectCorrelationInformation(event, Collections.singletonList(event2), streamlineComponent); values = Collections.singletonList(event); expectMessage = String.format("%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s", timestampStr, StreamlineEventLogger.DELIMITER, StreamlineEventLogger.MARKER_FOR_STREAMLINE_EVENT, StreamlineEventLogger.DELIMITER, streamlineComponent, StreamlineEventLogger.DELIMITER, event.getId(), StreamlineEventLogger.DELIMITER, "[" + event3.getId() + "]", StreamlineEventLogger.DELIMITER, "[" + event2.getId() + "]", StreamlineEventLogger.DELIMITER, ImmutableMap.copyOf(event), StreamlineEventLogger.DELIMITER, event.getHeader(), StreamlineEventLogger.DELIMITER, event.getAuxiliaryFieldsAndValues()); eventInfo = new IEventLogger.EventInfo(timestamp, stormComponentName, taskId, messageId, values); actualMessage = sut.buildLogMessage(eventInfo); Assert.assertEquals(expectMessage, actualMessage); }
@Test public void buildOtherEventLogMessage() { long timestamp = 1513300663123L; String timestampStr = dateFormat.format(timestamp); String stormComponentName = "1-Component"; int taskId = 2; Object messageId = "dummy"; List<Object> values = new ArrayList<>(); values.add("hello"); values.add("world"); values.add(12345); String expectMessage = String.format("%s,%s,%s,%s,%s,%s", timestampStr, StreamlineEventLogger.MARKER_FOR_OTHER_EVENT, stormComponentName, taskId, messageId, values); IEventLogger.EventInfo eventInfo = new IEventLogger.EventInfo(timestamp, stormComponentName, taskId, messageId, values); String actualMessage = sut.buildLogMessage(eventInfo); Assert.assertEquals(expectMessage, actualMessage); }