simple_kafka_vertx.java

class KafkaVerticle extends AbstractVerticle {

    // initialization

    @Override
    public void start() {
        KafkaConsumer.create(vertx, kafkaConfig)
                .subscribe(topic)
                .handler(record -> 
                    // processing logic comes here
                )
                .endHandler(v -> log.info("End of data. Topic: {}", this.topic))
                .exceptionHandler(e -> log.error("Single Kafka consumer error", e));
    }
}