class KafkaWriter extends StreamingProcessor {

    @Override
    protected EventSet consume(ProcessingContext processingContext, EventSet eventSet) throws ProcessingException {
        final List<KeyedMessage<String, String>> messages = Lists.newArrayList();
        try {
            eventSet.getEvents().forEach(event -> {
                KeyedMessage<String, String> convertedMessage = null;
                try {
                    convertedMessage = convertEvent(event);
                } catch (ProcessingException e) {
                    LOGGER.error("Error converting byte stream to event: ", e);
                    throw new RuntimeException(e);
                }
                if (null != convertedMessage) {
                    messages.add(convertedMessage);
                }
            });
        } catch (final Exception e) {
            LOGGER.error("Error converting byte stream to event: ", e);
            throw new ProcessingException(e);
        }
        Lists.partition(messages, ingestionPoolSize).forEach(messageList -> getProducer().send(messageList));
        return eventSet;
    }

}