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

                // Date, Marker, Component Name (Streamline), Event ID, Root IDs, Parent IDs,
                // Event Fields, Header KV, Aux. Fields KV
                // use DELIMITER to let parser understand it more easily
                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());
            }
        }

        // Date, Marker, Component Name (Storm), task ID, Message ID, Values
        // use comma-separated delimiter since this is not for machine, but for users
        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);
    }

}