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(); }); } CompositeFuture.all(futures).setHandler(event -> { kafkaProducers.clear(); future.complete(); }); return future; } KafkaProducerRepository(Vertx vertx);   }
@Test public void closeAll(TestContext context) { Async async = context.async(); repository.addKafkaProducer(new KafkaConfiguration(patternFrom("my.topic.*"), configs)); repository.addKafkaProducer(new KafkaConfiguration(patternFrom("my.other.topic.*"), configs_2)); context.assertTrue(repository.findMatchingKafkaProducer("my.topic.zz").isPresent()); context.assertTrue(repository.findMatchingKafkaProducer("my.other.topic.zz").isPresent()); repository.closeAll().setHandler(event -> { context.assertFalse(repository.findMatchingKafkaProducer("my.topic.zz").isPresent()); context.assertFalse(repository.findMatchingKafkaProducer("my.other.topic.zz").isPresent()); async.complete(); }); }