class KafkaProducerRepository {

    Future<Void> closeAll(){
        log.info("About to close all kafka producers");
        Future<Void> future = Future.future();
        List<Future> futures = new ArrayList<>();

        for (Map.Entry<Pattern, KafkaProducer<String, String>> entry : kafkaProducers.entrySet()) {
            Future entryFuture = Future.future();
            futures.add(entryFuture);
            entry.getValue().close(event -> {
                if(event.succeeded()){
                    log.info("Successfully closed producer for topic '{}'", entry.getKey().pattern());
                } else {
                    log.warn("Failed to close producer for topic '{}'", entry.getKey().pattern());
                }
                entryFuture.complete();
            });
        }

        // wait for all producers to be closed
        CompositeFuture.all(futures).setHandler(event -> {
            kafkaProducers.clear();
            future.complete();
        });
        return future;
    }

    public  KafkaProducerRepository(Vertx vertx);

}