KafkaHandler extends ConfigurationResourceConsumer { public Future<Void> initialize() { Future<Void> future = Future.future(); configurationResourceManager().getRegisteredResource(configResourceUri()).setHandler(event -> { if (event.succeeded() && event.result().isPresent()) { initializeKafkaConfiguration(event.result().get()).setHandler(event1 -> future.complete()); } else { log.warn("No kafka configuration resource with uri '{}' found. Unable to setup kafka configuration correctly", configResourceUri()); future.complete(); } }); return future; } KafkaHandler(ConfigurationResourceManager configurationResourceManager, KafkaProducerRepository repository,
                        KafkaMessageSender kafkaMessageSender, String configResourceUri, String streamingPath);  KafkaHandler(ConfigurationResourceManager configurationResourceManager, KafkaProducerRepository repository,
    KafkaMessageSender kafkaMessageSender, String configResourceUri, String streamingPath, Map<String, Object> properties); Future<Void> initialize(); boolean isInitialized(); boolean handle(final HttpServerRequest request); @Override void resourceChanged(String resourceUri, Buffer resource); @Override void resourceRemoved(String resourceUri);  }
@Test public void resourceRemovedTriggersCloseAllProducers(TestContext context){ Async async = context.async(); handler.initialize().setHandler(event -> { JsonObject object = new JsonObject(); object.put("requestUri", configResourceUri); object.put("type", "remove"); vertx.eventBus().publish(CONFIG_RESOURCE_CHANGED_ADDRESS, object); verify(repository, timeout(100).times(1)).closeAll(); verifyZeroInteractions(kafkaMessageSender); async.complete(); }); }