partitioned_kafka_vertx.java

// create required number of vertices
IntStream.range(0, numberOfPartitions)
    .forEach(partition -> {
        vertx.deployVerticle(
                () -> KafkaPartitionedVerticle.create(topic, partition, kafkaConfig),
                deploymentOptions,
                async -> log.info("Partitioned Kafka consumer deployed. DeploymentId: {}", async.result())
        );
    });


// inside KafkaPartitionedVerticle
class KafkaPartitionedVerticle extends AbstractVerticle {

    // initialization

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