class KafkaAdminClient implements Closeable {

    ZkUtils getZkUtils() {
        if (zkUtils == null) {
            Tuple2<ZkClient, ZkConnection> tuple;
            try {
                ZKConfig zkConfig = new ZKConfig(new VerifiableProperties(properties));
                tuple = ZkUtils.createZkClientAndConnection(zkConfig.zkConnect(), zkConfig.zkSessionTimeoutMs(),
                    zkConfig.zkConnectionTimeoutMs());
            } catch (ZkException | ZooKeeperClientException e) {
                throw new AdminOperationException("Unable to create admin connection", e);
            }

            boolean isSecure = Boolean.valueOf(properties.getProperty(ZOOKEEPER_SECURE, DEFAULT_ZOOKEEPER_SECURE));
            zkUtils = new ZkUtils(tuple._1(), tuple._2(), isSecure);
        }

        return zkUtils;
    }

    public  KafkaAdminClient(Properties properties);

}