class KafkaMessageSender {

    Future<Void> sendMessages(KafkaProducer<String, String> kafkaProducer,
                                     List<KafkaProducerRecord<String, String>> messages){
        Future<Void> future = Future.future();
        Future<Void> f = Future.succeededFuture();

        messages.stream().reduce(f,
                (f1, message) -> f1.compose(ignore -> KafkaMessageSender.this.sendMessage(kafkaProducer, message)),
                (voidFuture, voidFuture2) -> null
        ).setHandler(res -> {
            if(res.succeeded()) {
                future.complete();
            } else {
                future.fail(res.cause());
            }
        });

        return future;
    }

}