class KafkaMetadataService implements AutoCloseable {

    public static KafkaMetadataService newInstance(
            EnvironmentService environmentService, Long clusterId, SecurityContext securityContext)
                throws ServiceConfigurationNotFoundException, IOException, ServiceNotFoundException, ServiceComponentNotFoundException {

        final KafkaZkConnection kafkaZkConnection = KafkaZkConnection.newInstance(
                getZkStringRaw(environmentService, clusterId, AMBARI_JSON_CONFIG_KAFKA_BROKER));
        final ZookeeperClient zkCli = ZookeeperClient.newInstance(kafkaZkConnection);
        zkCli.start();

        return new KafkaMetadataService(zkCli, kafkaZkConnection, securityContext,
                getKafkaBrokerComponent(environmentService, clusterId),
                getKafkaBrokers(environmentService, clusterId),
                getServiceConfig(environmentService, clusterId, AMBARI_JSON_CONFIG_KAFKA_BROKER),
                getServiceConfig(environmentService, clusterId, AMBARI_JSON_CONFIG_KAFKA_ENV));
    }

      KafkaMetadataService(ZookeeperClient zkCli, KafkaZkConnection kafkaZkConnection,
                         SecurityContext securityContext, Component kafkaBroker, Collection<ComponentProcess> kafkaBrokerProcesses,
                         ServiceConfiguration brokerConfig, ServiceConfiguration kafkaEnvConfig);

}