@Test
public void resourceChangedTriggersCloseAllAndReCreateOfProducers(TestContext context){

    Async async = context.async();
    context.assertFalse(handler.isInitialized());
    storage.putMockData(configResourceUri, CONFIG_RESOURCE);
    JsonObject object = new JsonObject();
    object.put("requestUri", configResourceUri);
    object.put("type", "change");
    vertx.eventBus().publish(CONFIG_RESOURCE_CHANGED_ADDRESS, object);
    verify(repository, timeout(500).times(1)).closeAll();
    Map<String, String> configs_1 = new HashMap<String, String>() {{
        put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        put("bootstrap.servers", "localhost:9092");
    }};
    verify(repository, timeout(500).times(1)).addKafkaProducer(eq(new KafkaConfiguration(Pattern.compile("my.topic.*"), configs_1)));
    Map<String, String> configs_2 = new HashMap<String, String>() {{
        put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        put("bootstrap.servers", "localhost:9093");
        put("acks", "1");
    }};
    verify(repository, timeout(500).times(1)).addKafkaProducer(eq(new KafkaConfiguration(Pattern.compile("."), configs_2)));
    verifyZeroInteractions(kafkaMessageSender);
    await().atMost(1, SECONDS).until( () -> handler.isInitialized(), equalTo(Boolean.TRUE));
    async.complete();
}