KafkaHandler extends ConfigurationResourceConsumer { public boolean isInitialized() { return initialized; } 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 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(); }