ProduceToKafka.java

@Override
public void onEvent(EventLog eventLog, long l, boolean b) throws Exception {
    String topic = eventLog.getEventType();
    String json = eventLog.getValue();

    // send event log to kafka.
    if (this.sendMessageEnabled) {
        producer.send(new ProducerRecord<Integer, String>(topic, json));
    }
}

.