class KafkaAdminClient implements Closeable {

    public void updateTopicConfig(String topic, Properties properties) {
        if (StringUtils.isBlank(topic))
            throw new IllegalArgumentException("topic cannot be null, empty or blank");
        if (properties == null)
            throw new IllegalArgumentException("properties cannot be null");

        LOG.debug("Updating topic config for topic [{}] with config [{}]", topic, properties);

        try {
            List<ConfigEntry> configEntries = new ArrayList<>();
            for (String property : properties.stringPropertyNames()) {
                configEntries.add(new ConfigEntry(property, properties.getProperty(property)));
            }

            getNewAdminClient()
                .alterConfigs(
                    Collections.singletonMap(
                        new ConfigResource(ConfigResource.Type.TOPIC, topic),
                        new Config(configEntries)))
                .all()
                .get(operationTimeout, TimeUnit.MILLISECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            throw new AdminOperationException("Unable to update configuration for topic: " + topic, e);
        }
    }

    public  KafkaAdminClient(Properties properties);

}