@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();
    });
}